mirror of
https://github.com/pezkuwichain/pezkuwi-subxt.git
synced 2026-06-11 21:11:07 +00:00
Introduce Prometheus metric endpoint replacing Grafana endpoint (#4981)
* Refactor rebase master prometheus_v0.3 * Milestone1: Final Version of v0.3 * no-std or warm compatibility issues, grapana-data -source code reference and correction,applicable * Cargo.lock paritytech/master rebase * prometheus networking.rs del, grafana-data-source networking.rs pub edit and note * chore: reflect various feedback * Spaces to tabs. * Replace grafana and tidy * Add generics * Add photo back * Re-fix spaces in primitives/consensus/babe/src/inherents.rs * Refactor rebase master prometheus_v0.3 * Milestone1: Final Version of v0.3 * no-std or warm compatibility issues, grapana-data -source code reference and correction,applicable * prometheus networking.rs del, grafana-data-source networking.rs pub edit and note * chore: reflect various feedback * Replace grafana and tidy * Add generics * Add photo back * Re-fix spaces in primitives/consensus/babe/src/inherents.rs * chore: revert this file back to paritytech/master inherents.rs. * Add newline at EOF * Tidy * Use local registry * fix typo Co-Authored-By: Max Inden <mail@max-inden.de> * chore: Apply review feedback * endpoint -> exporter * fix readme * Remove lazy_static, use ServiceMetrics struct instead * Switch to using GaugeVecs * chore: without nightly , edit README * block_height -> block_height_number * Switch to a ready_transactions_number gauge * Update utils/prometheus/src/lib.rs Co-Authored-By: Max Inden <mail@max-inden.de> * no-prometheus flag add * /metrics url Input check * remove prometheus in Tracing * remove prometheus in Tracing * chore: master code rebase edit * gitlab-check-web-wasm edit code * From:from and cargo.lock update * with_prometheus_registry add background_tasks * utils/prometheus/src/lib.rs: Restructure #[cfg] for wasm without hyper Given that Hyper is not compatible with WASM targets it needs to be excluded from WASM builds. Instead of introducing #[cfg] lines throughout the crate, this patch splits the crate into two: known_os and unknown_os (WASM). * utils/prometheus/src/lib.rs: Feature gate known_os module * client/cli/src/lib.rs: Re-add newline at end of file Co-authored-by: JeseonLEE <zeroday26@gmail.com> Co-authored-by: Gavin Wood <github@gavwood.com> Co-authored-by: Ashley <ashley.ruglys@gmail.com> Co-authored-by: Hyungsuk Kang <hskang9@gmail.com>
This commit is contained in:
Max Inden
Generated
+98
-49
@@ -446,6 +446,12 @@ dependencies = [
|
||||
"wasm-bindgen-futures",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "bs58"
|
||||
version = "0.2.5"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "c95ee6bba9d950218b6cc910cf62bc9e0a171d0f4537e3627b0f54d08549b188"
|
||||
|
||||
[[package]]
|
||||
name = "bs58"
|
||||
version = "0.3.0"
|
||||
@@ -610,7 +616,6 @@ dependencies = [
|
||||
"js-sys",
|
||||
"num-integer",
|
||||
"num-traits",
|
||||
"serde",
|
||||
"time",
|
||||
"wasm-bindgen",
|
||||
]
|
||||
@@ -1952,34 +1957,6 @@ dependencies = [
|
||||
"scroll",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "grafana-data-source"
|
||||
version = "0.8.0"
|
||||
dependencies = [
|
||||
"async-std",
|
||||
"chrono",
|
||||
"derive_more",
|
||||
"futures-timer 3.0.1",
|
||||
"futures-util",
|
||||
"hyper 0.13.2",
|
||||
"lazy_static",
|
||||
"log 0.4.8",
|
||||
"parking_lot 0.10.0",
|
||||
"serde",
|
||||
"serde_json",
|
||||
"tokio 0.2.11",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "grafana-data-source-test"
|
||||
version = "2.0.0"
|
||||
dependencies = [
|
||||
"futures 0.3.4",
|
||||
"futures-timer 3.0.1",
|
||||
"grafana-data-source",
|
||||
"rand 0.7.3",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "h2"
|
||||
version = "0.1.26"
|
||||
@@ -2687,8 +2664,8 @@ dependencies = [
|
||||
"libp2p-wasm-ext",
|
||||
"libp2p-websocket",
|
||||
"libp2p-yamux",
|
||||
"parity-multiaddr",
|
||||
"parity-multihash",
|
||||
"parity-multiaddr 0.7.2",
|
||||
"parity-multihash 0.2.3",
|
||||
"parking_lot 0.10.0",
|
||||
"pin-project",
|
||||
"smallvec 1.2.0",
|
||||
@@ -2702,7 +2679,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "3b874594c4b29de1a29f27871feba8e6cd13aa54a8a1e8f8c7cf3dfac5ca287c"
|
||||
dependencies = [
|
||||
"asn1_der",
|
||||
"bs58",
|
||||
"bs58 0.3.0",
|
||||
"ed25519-dalek",
|
||||
"fnv",
|
||||
"futures 0.3.4",
|
||||
@@ -2711,8 +2688,8 @@ dependencies = [
|
||||
"libsecp256k1",
|
||||
"log 0.4.8",
|
||||
"multistream-select",
|
||||
"parity-multiaddr",
|
||||
"parity-multihash",
|
||||
"parity-multiaddr 0.7.2",
|
||||
"parity-multihash 0.2.3",
|
||||
"parking_lot 0.10.0",
|
||||
"pin-project",
|
||||
"prost",
|
||||
@@ -2723,7 +2700,7 @@ dependencies = [
|
||||
"sha2",
|
||||
"smallvec 1.2.0",
|
||||
"thiserror",
|
||||
"unsigned-varint",
|
||||
"unsigned-varint 0.3.0",
|
||||
"void",
|
||||
"zeroize 1.1.0",
|
||||
]
|
||||
@@ -2798,7 +2775,7 @@ dependencies = [
|
||||
"rand 0.7.3",
|
||||
"sha2",
|
||||
"smallvec 1.2.0",
|
||||
"unsigned-varint",
|
||||
"unsigned-varint 0.3.0",
|
||||
"wasm-timer",
|
||||
]
|
||||
|
||||
@@ -2833,14 +2810,14 @@ dependencies = [
|
||||
"libp2p-core",
|
||||
"libp2p-swarm",
|
||||
"log 0.4.8",
|
||||
"parity-multihash",
|
||||
"parity-multihash 0.2.3",
|
||||
"prost",
|
||||
"prost-build",
|
||||
"rand 0.7.3",
|
||||
"sha2",
|
||||
"smallvec 1.2.0",
|
||||
"uint",
|
||||
"unsigned-varint",
|
||||
"unsigned-varint 0.3.0",
|
||||
"void",
|
||||
"wasm-timer",
|
||||
]
|
||||
@@ -2880,7 +2857,7 @@ dependencies = [
|
||||
"libp2p-core",
|
||||
"log 0.4.8",
|
||||
"parking_lot 0.10.0",
|
||||
"unsigned-varint",
|
||||
"unsigned-varint 0.3.0",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
@@ -2933,7 +2910,7 @@ dependencies = [
|
||||
"prost",
|
||||
"prost-build",
|
||||
"rw-stream-sink",
|
||||
"unsigned-varint",
|
||||
"unsigned-varint 0.3.0",
|
||||
"void",
|
||||
]
|
||||
|
||||
@@ -3334,7 +3311,7 @@ dependencies = [
|
||||
"log 0.4.8",
|
||||
"smallvec 1.2.0",
|
||||
"tokio-io",
|
||||
"unsigned-varint",
|
||||
"unsigned-varint 0.3.0",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
@@ -4576,6 +4553,24 @@ version = "0.1.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "0c276d76c5333b8c2579e02d49a06733a55b8282d2d9b13e8d53b6406bd7e30a"
|
||||
|
||||
[[package]]
|
||||
name = "parity-multiaddr"
|
||||
version = "0.5.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "045b3c7af871285146300da35b1932bb6e4639b66c7c98e85d06a32cbc4e8fa7"
|
||||
dependencies = [
|
||||
"arrayref",
|
||||
"bs58 0.2.5",
|
||||
"byteorder 1.3.4",
|
||||
"bytes 0.4.12",
|
||||
"data-encoding",
|
||||
"parity-multihash 0.1.3",
|
||||
"percent-encoding 1.0.1",
|
||||
"serde",
|
||||
"unsigned-varint 0.2.3",
|
||||
"url 1.7.2",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "parity-multiaddr"
|
||||
version = "0.7.2"
|
||||
@@ -4583,17 +4578,32 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "26df883298bc3f4e92528b4c5cc9f806b791955b136da3e5e939ed9de0fd958b"
|
||||
dependencies = [
|
||||
"arrayref",
|
||||
"bs58",
|
||||
"bs58 0.3.0",
|
||||
"byteorder 1.3.4",
|
||||
"data-encoding",
|
||||
"parity-multihash",
|
||||
"parity-multihash 0.2.3",
|
||||
"percent-encoding 2.1.0",
|
||||
"serde",
|
||||
"static_assertions",
|
||||
"unsigned-varint",
|
||||
"unsigned-varint 0.3.0",
|
||||
"url 2.1.1",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "parity-multihash"
|
||||
version = "0.1.3"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "df3a17dc27848fd99e4f87eb0f8c9baba6ede0a6d555400c850ca45254ef4ce3"
|
||||
dependencies = [
|
||||
"blake2",
|
||||
"bytes 0.4.12",
|
||||
"rand 0.6.5",
|
||||
"sha-1",
|
||||
"sha2",
|
||||
"sha3",
|
||||
"unsigned-varint 0.2.3",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "parity-multihash"
|
||||
version = "0.2.3"
|
||||
@@ -4606,7 +4616,7 @@ dependencies = [
|
||||
"sha-1",
|
||||
"sha2",
|
||||
"sha3",
|
||||
"unsigned-varint",
|
||||
"unsigned-varint 0.3.0",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
@@ -4980,6 +4990,33 @@ dependencies = [
|
||||
"unicode-xid 0.2.0",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "prometheus"
|
||||
version = "0.7.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "5567486d5778e2c6455b1b90ff1c558f29e751fc018130fa182e15828e728af1"
|
||||
dependencies = [
|
||||
"cfg-if",
|
||||
"fnv",
|
||||
"lazy_static",
|
||||
"protobuf",
|
||||
"quick-error",
|
||||
"spin",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "prometheus-exporter"
|
||||
version = "0.8.0"
|
||||
dependencies = [
|
||||
"async-std",
|
||||
"derive_more",
|
||||
"futures-util",
|
||||
"hyper 0.13.2",
|
||||
"log 0.4.8",
|
||||
"prometheus",
|
||||
"tokio 0.2.11",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "prost"
|
||||
version = "0.6.1"
|
||||
@@ -5031,6 +5068,12 @@ dependencies = [
|
||||
"prost",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "protobuf"
|
||||
version = "2.10.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "6686ddd96a8dbe2687b5f2a687b2cfb520854010ec480f2d74c32e7c9873d3c5"
|
||||
|
||||
[[package]]
|
||||
name = "pwasm-utils"
|
||||
version = "0.12.0"
|
||||
@@ -5668,6 +5711,7 @@ dependencies = [
|
||||
"log 0.4.8",
|
||||
"names",
|
||||
"parity-util-mem",
|
||||
"prometheus-exporter",
|
||||
"regex",
|
||||
"rpassword",
|
||||
"sc-client-api",
|
||||
@@ -6201,7 +6245,7 @@ dependencies = [
|
||||
"substrate-test-runtime-client",
|
||||
"tempfile",
|
||||
"thiserror",
|
||||
"unsigned-varint",
|
||||
"unsigned-varint 0.3.0",
|
||||
"void",
|
||||
"wasm-timer",
|
||||
"zeroize 1.1.0",
|
||||
@@ -6389,13 +6433,13 @@ dependencies = [
|
||||
"futures 0.3.4",
|
||||
"futures-diagnose",
|
||||
"futures-timer 3.0.1",
|
||||
"grafana-data-source",
|
||||
"lazy_static",
|
||||
"log 0.4.8",
|
||||
"parity-multiaddr",
|
||||
"parity-multiaddr 0.5.0",
|
||||
"parity-scale-codec",
|
||||
"parity-util-mem",
|
||||
"parking_lot 0.10.0",
|
||||
"prometheus-exporter",
|
||||
"sc-chain-spec",
|
||||
"sc-client",
|
||||
"sc-client-api",
|
||||
@@ -6489,7 +6533,6 @@ name = "sc-tracing"
|
||||
version = "2.0.0"
|
||||
dependencies = [
|
||||
"erased-serde",
|
||||
"grafana-data-source",
|
||||
"log 0.4.8",
|
||||
"parking_lot 0.10.0",
|
||||
"sc-telemetry",
|
||||
@@ -8496,6 +8539,12 @@ version = "0.2.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "826e7639553986605ec5979c7dd957c7895e93eabed50ab2ffa7f6128a75097c"
|
||||
|
||||
[[package]]
|
||||
name = "unsigned-varint"
|
||||
version = "0.2.3"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "a7f0023a96687fe169081e8adce3f65e3874426b7886e9234d490af2dc077959"
|
||||
|
||||
[[package]]
|
||||
name = "unsigned-varint"
|
||||
version = "0.3.0"
|
||||
|
||||
@@ -53,9 +53,8 @@ members = [
|
||||
"client/telemetry",
|
||||
"client/transaction-pool",
|
||||
"client/transaction-pool/graph",
|
||||
"utils/prometheus",
|
||||
"utils/wasm-builder-runner",
|
||||
"utils/grafana-data-source",
|
||||
"utils/grafana-data-source/test",
|
||||
"frame/assets",
|
||||
"frame/aura",
|
||||
"frame/authority-discovery",
|
||||
|
||||
@@ -31,6 +31,7 @@ sp-core = { version = "2.0.0", path = "../../primitives/core" }
|
||||
sc-service = { version = "0.8", default-features = false, path = "../service" }
|
||||
sp-state-machine = { version = "0.8", path = "../../primitives/state-machine" }
|
||||
sc-telemetry = { version = "2.0.0", path = "../telemetry" }
|
||||
prometheus-exporter = { path = "../../utils/prometheus" }
|
||||
sp-keyring = { version = "2.0.0", path = "../../primitives/keyring" }
|
||||
names = "0.11.0"
|
||||
structopt = "0.3.8"
|
||||
|
||||
@@ -623,13 +623,6 @@ where
|
||||
config.rpc_ws = Some(parse_address(&format!("{}:{}", ws_interface, 9944), cli.ws_port)?);
|
||||
}
|
||||
|
||||
if config.grafana_port.is_none() || cli.grafana_port.is_some() {
|
||||
let grafana_interface: &str = if cli.grafana_external { "0.0.0.0" } else { "127.0.0.1" };
|
||||
config.grafana_port = Some(
|
||||
parse_address(&format!("{}:{}", grafana_interface, 9955), cli.grafana_port)?
|
||||
);
|
||||
}
|
||||
|
||||
config.rpc_ws_max_connections = cli.ws_max_connections;
|
||||
config.rpc_cors = cli.rpc_cors.unwrap_or_else(|| if is_dev {
|
||||
log::warn!("Running in --dev mode, RPC CORS has been disabled.");
|
||||
@@ -651,6 +644,14 @@ where
|
||||
} else if !cli.telemetry_endpoints.is_empty() {
|
||||
config.telemetry_endpoints = Some(TelemetryEndpoints::new(cli.telemetry_endpoints));
|
||||
}
|
||||
// Override prometheus
|
||||
if cli.no_prometheus {
|
||||
config.prometheus_port = None;
|
||||
} else {
|
||||
let prometheus_interface: &str = if cli.prometheus_external { "0.0.0.0" } else { "127.0.0.1" };
|
||||
config.prometheus_port = Some(
|
||||
parse_address(&format!("{}:{}", prometheus_interface, 9615), cli.prometheus_port)?);
|
||||
}
|
||||
|
||||
config.tracing_targets = cli.import_params.tracing_targets.into();
|
||||
config.tracing_receiver = cli.import_params.tracing_receiver.into();
|
||||
|
||||
@@ -337,7 +337,6 @@ arg_enum! {
|
||||
pub enum TracingReceiver {
|
||||
Log,
|
||||
Telemetry,
|
||||
Grafana,
|
||||
}
|
||||
}
|
||||
|
||||
@@ -346,7 +345,6 @@ impl Into<sc_tracing::TracingReceiver> for TracingReceiver {
|
||||
match self {
|
||||
TracingReceiver::Log => sc_tracing::TracingReceiver::Log,
|
||||
TracingReceiver::Telemetry => sc_tracing::TracingReceiver::Telemetry,
|
||||
TracingReceiver::Grafana => sc_tracing::TracingReceiver::Grafana,
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -486,11 +484,11 @@ pub struct RunCmd {
|
||||
#[structopt(long = "unsafe-ws-external")]
|
||||
pub unsafe_ws_external: bool,
|
||||
|
||||
/// Listen to all Grafana data source interfaces.
|
||||
/// Listen to all Prometheus endpoint interfaces.
|
||||
///
|
||||
/// Default is local.
|
||||
#[structopt(long = "grafana-external")]
|
||||
pub grafana_external: bool,
|
||||
#[structopt(long = "prometheus-external")]
|
||||
pub prometheus_external: bool,
|
||||
|
||||
/// Specify HTTP RPC server TCP port.
|
||||
#[structopt(long = "rpc-port", value_name = "PORT")]
|
||||
@@ -514,9 +512,15 @@ pub struct RunCmd {
|
||||
#[structopt(long = "rpc-cors", value_name = "ORIGINS", parse(try_from_str = parse_cors))]
|
||||
pub rpc_cors: Option<Cors>,
|
||||
|
||||
/// Specify Grafana data source server TCP Port.
|
||||
#[structopt(long = "grafana-port", value_name = "PORT")]
|
||||
pub grafana_port: Option<u16>,
|
||||
/// Specify Prometheus endpoint TCP Port.
|
||||
#[structopt(long = "prometheus-port", value_name = "PORT")]
|
||||
pub prometheus_port: Option<u16>,
|
||||
|
||||
/// Do not expose a Prometheus metric endpoint.
|
||||
///
|
||||
/// Prometheus metric endpoint is enabled by default.
|
||||
#[structopt(long = "no-prometheus")]
|
||||
pub no_prometheus: bool,
|
||||
|
||||
/// The human-readable name for this node.
|
||||
///
|
||||
|
||||
@@ -52,8 +52,8 @@ sc-rpc-server = { version = "2.0.0", path = "../rpc-servers" }
|
||||
sc-rpc = { version = "2.0.0", path = "../rpc" }
|
||||
sc-telemetry = { version = "2.0.0", path = "../telemetry" }
|
||||
sc-offchain = { version = "2.0.0", path = "../offchain" }
|
||||
parity-multiaddr = { package = "parity-multiaddr", version = "0.7.1" }
|
||||
grafana-data-source = { version = "0.8", path = "../../utils/grafana-data-source" }
|
||||
parity-multiaddr = { package = "parity-multiaddr", version = "0.5.0" }
|
||||
prometheus-exporter = { path = "../../utils/prometheus" }
|
||||
sc-tracing = { version = "2.0.0", path = "../tracing" }
|
||||
tracing = "0.1.10"
|
||||
parity-util-mem = { version = "0.5.1", default-features = false, features = ["primitive-types"] }
|
||||
|
||||
@@ -39,7 +39,7 @@ use sc_network::{config::BoxFinalityProofRequestBuilder, specialization::Network
|
||||
use parking_lot::{Mutex, RwLock};
|
||||
use sp_runtime::generic::BlockId;
|
||||
use sp_runtime::traits::{
|
||||
Block as BlockT, NumberFor, SaturatedConversion, HasherFor,
|
||||
Block as BlockT, NumberFor, SaturatedConversion, HasherFor, UniqueSaturatedInto,
|
||||
};
|
||||
use sp_api::ProvideRuntimeApi;
|
||||
use sc_executor::{NativeExecutor, NativeExecutionDispatch};
|
||||
@@ -53,7 +53,43 @@ use sysinfo::{get_current_pid, ProcessExt, System, SystemExt};
|
||||
use sc_telemetry::{telemetry, SUBSTRATE_INFO};
|
||||
use sp_transaction_pool::{MaintainedTransactionPool, ChainEvent};
|
||||
use sp_blockchain;
|
||||
use grafana_data_source::{self, record_metrics};
|
||||
use prometheus_exporter::{register, Gauge, U64, F64, Registry, PrometheusError, Opts, GaugeVec};
|
||||
|
||||
struct ServiceMetrics {
|
||||
block_height_number: GaugeVec<U64>,
|
||||
peers_count: Gauge<U64>,
|
||||
ready_transactions_number: Gauge<U64>,
|
||||
memory_usage_bytes: Gauge<U64>,
|
||||
cpu_usage_percentage: Gauge<F64>,
|
||||
network_per_sec_bytes: GaugeVec<U64>,
|
||||
}
|
||||
|
||||
impl ServiceMetrics {
|
||||
fn register(registry: &Registry) -> Result<Self, PrometheusError> {
|
||||
Ok(Self {
|
||||
block_height_number: register(GaugeVec::new(
|
||||
Opts::new("block_height_number", "Height of the chain"),
|
||||
&["status"]
|
||||
)?, registry)?,
|
||||
peers_count: register(Gauge::new(
|
||||
"peers_count", "Number of network gossip peers",
|
||||
)?, registry)?,
|
||||
ready_transactions_number: register(Gauge::new(
|
||||
"ready_transactions_number", "Number of transactions in the ready queue",
|
||||
)?, registry)?,
|
||||
memory_usage_bytes: register(Gauge::new(
|
||||
"memory_usage_bytes", "Node memory usage",
|
||||
)?, registry)?,
|
||||
cpu_usage_percentage: register(Gauge::new(
|
||||
"cpu_usage_percentage", "Node CPU usage",
|
||||
)?, registry)?,
|
||||
network_per_sec_bytes: register(GaugeVec::new(
|
||||
Opts::new("network_per_sec_bytes", "Networking bytes per second"),
|
||||
&["direction"]
|
||||
)?, registry)?,
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
pub type BackgroundTask = Pin<Box<dyn Future<Output=()> + Send>>;
|
||||
|
||||
@@ -93,6 +129,7 @@ pub struct ServiceBuilder<TBl, TRtApi, TGen, TCSExt, TCl, TFchr, TSc, TImpQu, TF
|
||||
remote_backend: Option<Arc<dyn RemoteBlockchain<TBl>>>,
|
||||
marker: PhantomData<(TBl, TRtApi)>,
|
||||
background_tasks: Vec<(&'static str, BackgroundTask)>,
|
||||
prometheus_registry: Option<Registry>
|
||||
}
|
||||
|
||||
/// Full client type.
|
||||
@@ -270,6 +307,7 @@ where TGen: RuntimeGenesis, TCSExt: Extension {
|
||||
remote_backend: None,
|
||||
background_tasks: Default::default(),
|
||||
marker: PhantomData,
|
||||
prometheus_registry: None,
|
||||
})
|
||||
}
|
||||
|
||||
@@ -356,6 +394,7 @@ where TGen: RuntimeGenesis, TCSExt: Extension {
|
||||
remote_backend: Some(remote_blockchain),
|
||||
background_tasks: Default::default(),
|
||||
marker: PhantomData,
|
||||
prometheus_registry: None,
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -429,6 +468,7 @@ impl<TBl, TRtApi, TGen, TCSExt, TCl, TFchr, TSc, TImpQu, TFprb, TFpp, TNetP, TEx
|
||||
remote_backend: self.remote_backend,
|
||||
background_tasks: self.background_tasks,
|
||||
marker: self.marker,
|
||||
prometheus_registry: self.prometheus_registry,
|
||||
})
|
||||
}
|
||||
|
||||
@@ -472,6 +512,7 @@ impl<TBl, TRtApi, TGen, TCSExt, TCl, TFchr, TSc, TImpQu, TFprb, TFpp, TNetP, TEx
|
||||
remote_backend: self.remote_backend,
|
||||
background_tasks: self.background_tasks,
|
||||
marker: self.marker,
|
||||
prometheus_registry: self.prometheus_registry,
|
||||
})
|
||||
}
|
||||
|
||||
@@ -499,6 +540,7 @@ impl<TBl, TRtApi, TGen, TCSExt, TCl, TFchr, TSc, TImpQu, TFprb, TFpp, TNetP, TEx
|
||||
remote_backend: self.remote_backend,
|
||||
background_tasks: self.background_tasks,
|
||||
marker: self.marker,
|
||||
prometheus_registry: self.prometheus_registry,
|
||||
})
|
||||
}
|
||||
|
||||
@@ -540,6 +582,7 @@ impl<TBl, TRtApi, TGen, TCSExt, TCl, TFchr, TSc, TImpQu, TFprb, TFpp, TNetP, TEx
|
||||
remote_backend: self.remote_backend,
|
||||
background_tasks: self.background_tasks,
|
||||
marker: self.marker,
|
||||
prometheus_registry: self.prometheus_registry,
|
||||
})
|
||||
}
|
||||
|
||||
@@ -605,6 +648,7 @@ impl<TBl, TRtApi, TGen, TCSExt, TCl, TFchr, TSc, TImpQu, TFprb, TFpp, TNetP, TEx
|
||||
remote_backend: self.remote_backend,
|
||||
background_tasks: self.background_tasks,
|
||||
marker: self.marker,
|
||||
prometheus_registry: self.prometheus_registry,
|
||||
})
|
||||
}
|
||||
|
||||
@@ -665,6 +709,7 @@ impl<TBl, TRtApi, TGen, TCSExt, TCl, TFchr, TSc, TImpQu, TFprb, TFpp, TNetP, TEx
|
||||
remote_backend: self.remote_backend,
|
||||
background_tasks: self.background_tasks,
|
||||
marker: self.marker,
|
||||
prometheus_registry: self.prometheus_registry,
|
||||
})
|
||||
}
|
||||
|
||||
@@ -693,8 +738,31 @@ impl<TBl, TRtApi, TGen, TCSExt, TCl, TFchr, TSc, TImpQu, TFprb, TFpp, TNetP, TEx
|
||||
remote_backend: self.remote_backend,
|
||||
background_tasks: self.background_tasks,
|
||||
marker: self.marker,
|
||||
prometheus_registry: self.prometheus_registry,
|
||||
})
|
||||
}
|
||||
|
||||
/// Use an existing prometheus `Registry` to record metrics into.
|
||||
pub fn with_prometheus_registry(self, registry: Registry) -> Self {
|
||||
Self {
|
||||
config: self.config,
|
||||
client: self.client,
|
||||
backend: self.backend,
|
||||
keystore: self.keystore,
|
||||
fetcher: self.fetcher,
|
||||
select_chain: self.select_chain,
|
||||
import_queue: self.import_queue,
|
||||
finality_proof_request_builder: self.finality_proof_request_builder,
|
||||
finality_proof_provider: self.finality_proof_provider,
|
||||
network_protocol: self.network_protocol,
|
||||
transaction_pool: self.transaction_pool,
|
||||
rpc_extensions: self.rpc_extensions,
|
||||
remote_backend: self.remote_backend,
|
||||
background_tasks: self.background_tasks,
|
||||
marker: self.marker,
|
||||
prometheus_registry: Some(registry),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Implemented on `ServiceBuilder`. Allows running block commands, such as import/export/validate
|
||||
@@ -807,6 +875,7 @@ ServiceBuilder<
|
||||
rpc_extensions,
|
||||
remote_backend,
|
||||
background_tasks,
|
||||
prometheus_registry,
|
||||
} = self;
|
||||
|
||||
sp_session::generate_initial_session_keys(
|
||||
@@ -998,6 +1067,30 @@ ServiceBuilder<
|
||||
));
|
||||
}
|
||||
|
||||
// Prometheus exporter and metrics
|
||||
let metrics = if let Some(port) = config.prometheus_port {
|
||||
let registry = match prometheus_registry {
|
||||
Some(registry) => registry,
|
||||
None => Registry::new_custom(Some("substrate".into()), None)?
|
||||
};
|
||||
|
||||
let metrics = ServiceMetrics::register(®istry)?;
|
||||
|
||||
let future = select(
|
||||
prometheus_exporter::init_prometheus(port, registry).boxed(),
|
||||
exit.clone()
|
||||
).map(drop);
|
||||
|
||||
let _ = to_spawn_tx.unbounded_send((
|
||||
Box::pin(future),
|
||||
From::from("prometheus-endpoint")
|
||||
));
|
||||
|
||||
Some(metrics)
|
||||
} else {
|
||||
None
|
||||
};
|
||||
|
||||
// Periodically notify the telemetry.
|
||||
let transaction_pool_ = transaction_pool.clone();
|
||||
let client_ = client.clone();
|
||||
@@ -1014,6 +1107,8 @@ ServiceBuilder<
|
||||
let finalized_number: u64 = info.chain.finalized_number.saturated_into::<u64>();
|
||||
let bandwidth_download = net_status.average_download_per_sec;
|
||||
let bandwidth_upload = net_status.average_upload_per_sec;
|
||||
let best_seen_block = net_status.best_seen_block
|
||||
.map(|num: NumberFor<TBl>| num.unique_saturated_into() as u64);
|
||||
|
||||
// get cpu usage and memory usage of this process
|
||||
let (cpu_usage, memory) = if let Some(self_pid) = self_pid {
|
||||
@@ -1042,25 +1137,22 @@ ServiceBuilder<
|
||||
"disk_read_per_sec" => info.usage.as_ref().map(|usage| usage.io.bytes_read).unwrap_or(0),
|
||||
"disk_write_per_sec" => info.usage.as_ref().map(|usage| usage.io.bytes_written).unwrap_or(0),
|
||||
);
|
||||
#[cfg(not(target_os = "unknown"))]
|
||||
let memory_transaction_pool = parity_util_mem::malloc_size(&*transaction_pool_);
|
||||
#[cfg(target_os = "unknown")]
|
||||
let memory_transaction_pool = 0;
|
||||
let _ = record_metrics!(
|
||||
"peers" => num_peers,
|
||||
"height" => best_number,
|
||||
"txcount" => txpool_status.ready,
|
||||
"cpu" => cpu_usage,
|
||||
"memory" => memory,
|
||||
"finalized_height" => finalized_number,
|
||||
"bandwidth_download" => bandwidth_download,
|
||||
"bandwidth_upload" => bandwidth_upload,
|
||||
"used_state_cache_size" => info.usage.as_ref().map(|usage| usage.memory.state_cache).unwrap_or(0),
|
||||
"used_db_cache_size" => info.usage.as_ref().map(|usage| usage.memory.database_cache).unwrap_or(0),
|
||||
"disk_read_per_sec" => info.usage.as_ref().map(|usage| usage.io.bytes_read).unwrap_or(0),
|
||||
"disk_write_per_sec" => info.usage.as_ref().map(|usage| usage.io.bytes_written).unwrap_or(0),
|
||||
"memory_transaction_pool" => memory_transaction_pool,
|
||||
);
|
||||
if let Some(metrics) = metrics.as_ref() {
|
||||
metrics.memory_usage_bytes.set(memory);
|
||||
metrics.cpu_usage_percentage.set(f64::from(cpu_usage));
|
||||
metrics.ready_transactions_number.set(txpool_status.ready as u64);
|
||||
metrics.peers_count.set(num_peers as u64);
|
||||
|
||||
metrics.network_per_sec_bytes.with_label_values(&["download"]).set(net_status.average_download_per_sec);
|
||||
metrics.network_per_sec_bytes.with_label_values(&["upload"]).set(net_status.average_upload_per_sec);
|
||||
|
||||
metrics.block_height_number.with_label_values(&["finalized"]).set(finalized_number);
|
||||
metrics.block_height_number.with_label_values(&["best"]).set(best_number);
|
||||
|
||||
if let Some(best_seen_block) = best_seen_block {
|
||||
metrics.block_height_number.with_label_values(&["sync_target"]).set(best_seen_block);
|
||||
}
|
||||
}
|
||||
|
||||
ready(())
|
||||
});
|
||||
@@ -1217,16 +1309,6 @@ ServiceBuilder<
|
||||
telemetry
|
||||
});
|
||||
|
||||
// Grafana data source
|
||||
if let Some(port) = config.grafana_port {
|
||||
let future = select(
|
||||
grafana_data_source::run_server(port).boxed(),
|
||||
exit.clone()
|
||||
).map(drop);
|
||||
|
||||
let _ = to_spawn_tx.unbounded_send((Box::pin(future), From::from("grafana-server")));
|
||||
}
|
||||
|
||||
// Instrumentation
|
||||
if let Some(tracing_targets) = config.tracing_targets.as_ref() {
|
||||
let subscriber = sc_tracing::ProfilingSubscriber::new(
|
||||
|
||||
@@ -93,8 +93,8 @@ pub struct Configuration<G, E = NoExtension> {
|
||||
pub rpc_ws_max_connections: Option<usize>,
|
||||
/// CORS settings for HTTP & WS servers. `None` if all origins are allowed.
|
||||
pub rpc_cors: Option<Vec<String>>,
|
||||
/// Grafana data source http port. `None` if disabled.
|
||||
pub grafana_port: Option<SocketAddr>,
|
||||
/// Prometheus exporter Port. `None` if disabled.
|
||||
pub prometheus_port: Option<SocketAddr>,
|
||||
/// Telemetry service URL. `None` if disabled.
|
||||
pub telemetry_endpoints: Option<TelemetryEndpoints>,
|
||||
/// External WASM transport for the telemetry. If `Some`, when connection to a telemetry
|
||||
@@ -190,7 +190,7 @@ impl<G, E> Default for Configuration<G, E> {
|
||||
rpc_ws: None,
|
||||
rpc_ws_max_connections: None,
|
||||
rpc_cors: Some(vec![]),
|
||||
grafana_port: None,
|
||||
prometheus_port: None,
|
||||
telemetry_endpoints: None,
|
||||
telemetry_external_transport: None,
|
||||
default_heap_pages: None,
|
||||
|
||||
@@ -53,6 +53,12 @@ impl<'a> From<&'a str> for Error {
|
||||
}
|
||||
}
|
||||
|
||||
impl From<prometheus_exporter::PrometheusError> for Error {
|
||||
fn from(e: prometheus_exporter::PrometheusError) -> Self {
|
||||
Error::Other(format!("Prometheus error: {}", e))
|
||||
}
|
||||
}
|
||||
|
||||
impl std::error::Error for Error {
|
||||
fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
|
||||
match self {
|
||||
|
||||
@@ -199,7 +199,7 @@ fn node_config<G, E: Clone> (
|
||||
rpc_ws: None,
|
||||
rpc_ws_max_connections: None,
|
||||
rpc_cors: None,
|
||||
grafana_port: None,
|
||||
prometheus_port: None,
|
||||
telemetry_endpoints: None,
|
||||
telemetry_external_transport: None,
|
||||
default_heap_pages: None,
|
||||
|
||||
@@ -15,7 +15,6 @@ slog = { version = "2.5.2", features = ["nested-values"] }
|
||||
tracing-core = "0.1.7"
|
||||
|
||||
sc-telemetry = { version = "2.0.0", path = "../telemetry" }
|
||||
grafana-data-source = { version = "0.8", path = "../../utils/grafana-data-source" }
|
||||
|
||||
[dev-dependencies]
|
||||
tracing = "0.1.10"
|
||||
|
||||
@@ -34,7 +34,7 @@
|
||||
//! let span = tracing::span!(tracing::Level::INFO, "my_span_name", my_number = 10, a_key = "a value");
|
||||
//! let _guard = span.enter();
|
||||
//! ```
|
||||
//! Currently we provide `Log` (default), `Telemetry` and `Grafana` variants for `Receiver`
|
||||
//! Currently we provide `Log` (default), `Telemetry` variants for `Receiver`
|
||||
|
||||
use std::collections::HashMap;
|
||||
use std::fmt;
|
||||
@@ -53,7 +53,6 @@ use tracing_core::{
|
||||
subscriber::Subscriber
|
||||
};
|
||||
|
||||
use grafana_data_source::{self, record_metrics};
|
||||
use sc_telemetry::{telemetry, SUBSTRATE_INFO};
|
||||
|
||||
/// Used to configure how to receive the metrics
|
||||
@@ -63,8 +62,6 @@ pub enum TracingReceiver {
|
||||
Log,
|
||||
/// Output to telemetry
|
||||
Telemetry,
|
||||
/// Output to Grafana
|
||||
Grafana,
|
||||
}
|
||||
|
||||
impl Default for TracingReceiver {
|
||||
@@ -255,7 +252,6 @@ impl ProfilingSubscriber {
|
||||
match self.receiver {
|
||||
TracingReceiver::Log => print_log(span_datum),
|
||||
TracingReceiver::Telemetry => send_telemetry(span_datum),
|
||||
TracingReceiver::Grafana => send_grafana(span_datum),
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -291,9 +287,3 @@ fn send_telemetry(span_datum: SpanDatum) {
|
||||
);
|
||||
}
|
||||
|
||||
fn send_grafana(span_datum: SpanDatum) {
|
||||
let name = format!("{}::{}", span_datum.target, span_datum.name);
|
||||
if let Err(e) = record_metrics!(&name => span_datum.overall_time.as_nanos(),) {
|
||||
log::warn!("Unable to send metrics to grafana: {:?}", e);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1,154 +0,0 @@
|
||||
// Copyright 2019-2020 Parity Technologies (UK) Ltd.
|
||||
// This file is part of Substrate.
|
||||
|
||||
// Substrate is free software: you can redistribute it and/or modify
|
||||
// it under the terms of the GNU General Public License as published by
|
||||
// the Free Software Foundation, either version 3 of the License, or
|
||||
// (at your option) any later version.
|
||||
|
||||
// Substrate is distributed in the hope that it will be useful,
|
||||
// but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||
// GNU General Public License for more details.
|
||||
|
||||
// You should have received a copy of the GNU General Public License
|
||||
// along with Substrate. If not, see <http://www.gnu.org/licenses/>.
|
||||
|
||||
use std::collections::HashMap;
|
||||
use std::convert::TryFrom;
|
||||
use crate::Error;
|
||||
|
||||
pub struct Database {
|
||||
base_timestamp: i64,
|
||||
storage: HashMap<String, Vec<Datapoint>>
|
||||
}
|
||||
|
||||
impl Database {
|
||||
/// Create a new Database.
|
||||
pub fn new() -> Self {
|
||||
Self {
|
||||
base_timestamp: now_millis(),
|
||||
storage: HashMap::new()
|
||||
}
|
||||
}
|
||||
|
||||
/// Produce an iterator for keys starting with a base string.
|
||||
pub fn keys_starting_with<'a>(&'a self, base: &'a str) -> impl Iterator<Item = String> + 'a {
|
||||
self.storage.keys()
|
||||
.filter(move |key| key.starts_with(base))
|
||||
.cloned()
|
||||
}
|
||||
|
||||
/// Select `max_datapoints` datapoints that have been added between `from` and `to`.
|
||||
pub fn datapoints_between(&self, key: &str, from: i64, to: i64, max_datapoints: usize) -> Option<Vec<(f32, i64)>> {
|
||||
self.storage.get(key)
|
||||
.map(|vec| {
|
||||
let from = find_index(vec, self.base_timestamp, from);
|
||||
let to = find_index(vec, self.base_timestamp, to);
|
||||
let slice = &vec[from .. to];
|
||||
|
||||
if max_datapoints == 0 {
|
||||
Vec::new()
|
||||
} else if max_datapoints >= slice.len() {
|
||||
// Just convert the slice as-is
|
||||
slice.iter()
|
||||
.map(|dp| dp.make_absolute(self.base_timestamp))
|
||||
.collect()
|
||||
} else {
|
||||
// We have more datapoints than we need, so we need to skip some
|
||||
(0 .. max_datapoints - 1)
|
||||
.map(|i| &slice[i * slice.len() / (max_datapoints - 1)])
|
||||
.chain(slice.last())
|
||||
.map(|dp| dp.make_absolute(self.base_timestamp))
|
||||
.collect()
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
/// Push a new datapoint. Will error if the base timestamp hasn't been updated in `2^32`
|
||||
/// milliseconds (49 days).
|
||||
pub fn push(&mut self, key: &str, value: f32) -> Result<(), Error> {
|
||||
self.storage.entry(key.into())
|
||||
.or_insert_with(Vec::new)
|
||||
.push(Datapoint::new(self.base_timestamp, value)?);
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Set a new base timestamp, and remove metrics older than this new timestamp. Errors if the
|
||||
/// difference between timestamps is greater than `2^32` milliseconds (49 days).
|
||||
pub fn truncate(&mut self, new_base_timestamp: i64) -> Result<(), Error> {
|
||||
// Ensure that the new base is older.
|
||||
if self.base_timestamp >= new_base_timestamp {
|
||||
return Ok(());
|
||||
}
|
||||
|
||||
// If the old base timestamp was too long ago, the
|
||||
let delta = u32::try_from(new_base_timestamp - self.base_timestamp)
|
||||
.map_err(Error::Timestamp)?;
|
||||
|
||||
for metric in self.storage.values_mut() {
|
||||
// Find the index of the oldest allowed timestamp and cut out all those before it.
|
||||
let index = find_index(&metric, self.base_timestamp, new_base_timestamp);
|
||||
|
||||
*metric = metric.iter_mut()
|
||||
.skip(index)
|
||||
.map(|dp| {
|
||||
dp.delta_timestamp -= delta;
|
||||
*dp
|
||||
})
|
||||
.collect();
|
||||
}
|
||||
|
||||
self.base_timestamp = new_base_timestamp;
|
||||
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Clone, Copy)]
|
||||
struct Datapoint {
|
||||
delta_timestamp: u32,
|
||||
value: f32
|
||||
}
|
||||
|
||||
impl Datapoint {
|
||||
fn new(base_timestamp: i64, value: f32) -> Result<Self, Error> {
|
||||
Ok(Self {
|
||||
delta_timestamp: u32::try_from(now_millis() - base_timestamp)
|
||||
.map_err(Error::Timestamp)?,
|
||||
value
|
||||
})
|
||||
}
|
||||
|
||||
fn make_absolute(self, base_timestamp: i64) -> (f32, i64) {
|
||||
(self.value, base_timestamp + self.delta_timestamp as i64)
|
||||
}
|
||||
}
|
||||
|
||||
fn find_index(slice: &[Datapoint], base_timestamp: i64, timestamp: i64) -> usize {
|
||||
slice.binary_search_by_key(×tamp, |datapoint| {
|
||||
base_timestamp + datapoint.delta_timestamp as i64
|
||||
}).unwrap_or_else(|index| index)
|
||||
}
|
||||
|
||||
/// Get the current unix timestamp in milliseconds.
|
||||
fn now_millis() -> i64 {
|
||||
chrono::Utc::now().timestamp_millis()
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test() {
|
||||
let mut database = Database::new();
|
||||
|
||||
database.push("test", 1.0).unwrap();
|
||||
database.push("test", 2.5).unwrap();
|
||||
database.push("test", 2.0).unwrap();
|
||||
database.push("test 2", 1.0).unwrap();
|
||||
|
||||
let mut keys: Vec<_> = database.keys_starting_with("test").collect();
|
||||
keys.sort();
|
||||
|
||||
assert_eq!(keys, ["test", "test 2"]);
|
||||
assert_eq!(database.keys_starting_with("test ").collect::<Vec<_>>(), ["test 2"]);
|
||||
}
|
||||
@@ -1,100 +0,0 @@
|
||||
// Copyright 2019-2020 Parity Technologies (UK) Ltd.
|
||||
// This file is part of Substrate.
|
||||
|
||||
// Substrate is free software: you can redistribute it and/or modify
|
||||
// it under the terms of the GNU General Public License as published by
|
||||
// the Free Software Foundation, either version 3 of the License, or
|
||||
// (at your option) any later version.
|
||||
|
||||
// Substrate is distributed in the hope that it will be useful,
|
||||
// but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||
// GNU General Public License for more details.
|
||||
|
||||
// You should have received a copy of the GNU General Public License
|
||||
// along with Substrate. If not, see <http://www.gnu.org/licenses/>.
|
||||
|
||||
//! [Grafana] data source server
|
||||
//!
|
||||
//! To display node statistics with [Grafana], this module exposes a `run_server` function that
|
||||
//! starts up a HTTP server that conforms to the [`grafana-json-data-source`] API. The
|
||||
//! `record_metrics` macro can be used to pass metrics to this server.
|
||||
//!
|
||||
//! [Grafana]: https://grafana.com/
|
||||
//! [`grafana-json-data-source`]: https://github.com/simPod/grafana-json-datasource
|
||||
|
||||
#![warn(missing_docs)]
|
||||
|
||||
use lazy_static::lazy_static;
|
||||
use parking_lot::RwLock;
|
||||
|
||||
mod types;
|
||||
mod server;
|
||||
#[cfg(not(target_os = "unknown"))]
|
||||
mod networking;
|
||||
mod database;
|
||||
|
||||
use database::Database;
|
||||
pub use server::run_server;
|
||||
use std::num::TryFromIntError;
|
||||
|
||||
lazy_static! {
|
||||
// The `RwLock` wrapping the metrics database.
|
||||
static ref DATABASE: RwLock<Database> = RwLock::new(Database::new());
|
||||
}
|
||||
|
||||
/// Write metrics to `METRICS`.
|
||||
#[macro_export]
|
||||
macro_rules! record_metrics(
|
||||
($($key:expr => $value:expr,)*) => {
|
||||
if cfg!(not(target_os = "unknown")) {
|
||||
$crate::record_metrics_slice(&[
|
||||
$( ($key, $value as f32), )*
|
||||
])
|
||||
} else {
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
);
|
||||
|
||||
/// Write metrics to `METRICS` as a slice. Intended to be only used via `record_metrics!`.
|
||||
pub fn record_metrics_slice(metrics: &[(&str, f32)]) -> Result<(), Error> {
|
||||
let mut database = crate::DATABASE.write();
|
||||
|
||||
for &(key, value) in metrics.iter() {
|
||||
database.push(key, value)?;
|
||||
}
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Error type that can be returned by either `record_metrics` or `run_server`.
|
||||
#[derive(Debug, derive_more::Display, derive_more::From)]
|
||||
pub enum Error {
|
||||
/// Hyper internal error.
|
||||
#[cfg(not(target_os = "unknown"))]
|
||||
Hyper(hyper::Error),
|
||||
/// Http request error.
|
||||
#[cfg(not(target_os = "unknown"))]
|
||||
Http(hyper::http::Error),
|
||||
/// Serialization/deserialization error.
|
||||
Serde(serde_json::Error),
|
||||
/// Timestamp error.
|
||||
Timestamp(TryFromIntError),
|
||||
/// i/o error.
|
||||
Io(std::io::Error)
|
||||
}
|
||||
|
||||
impl std::error::Error for Error {
|
||||
fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
|
||||
match self {
|
||||
#[cfg(not(target_os = "unknown"))]
|
||||
Error::Hyper(error) => Some(error),
|
||||
#[cfg(not(target_os = "unknown"))]
|
||||
Error::Http(error) => Some(error),
|
||||
Error::Serde(error) => Some(error),
|
||||
Error::Timestamp(error) => Some(error),
|
||||
Error::Io(error) => Some(error)
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -1,164 +0,0 @@
|
||||
// Copyright 2019-2020 Parity Technologies (UK) Ltd.
|
||||
// This file is part of Substrate.
|
||||
|
||||
// Substrate is free software: you can redistribute it and/or modify
|
||||
// it under the terms of the GNU General Public License as published by
|
||||
// the Free Software Foundation, either version 3 of the License, or
|
||||
// (at your option) any later version.
|
||||
|
||||
// Substrate is distributed in the hope that it will be useful,
|
||||
// but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||
// GNU General Public License for more details.
|
||||
|
||||
// You should have received a copy of the GNU General Public License
|
||||
// along with Substrate. If not, see <http://www.gnu.org/licenses/>.
|
||||
|
||||
use serde::{Serialize, de::DeserializeOwned};
|
||||
use chrono::{Duration, Utc};
|
||||
use futures_util::{FutureExt, TryStreamExt, future::{Future, select, Either}};
|
||||
use futures_timer::Delay;
|
||||
use crate::{DATABASE, Error, types::{Target, Query, TimeseriesData, Range}};
|
||||
|
||||
#[cfg(not(target_os = "unknown"))]
|
||||
use hyper::{Body, Request, Response, header, service::{service_fn, make_service_fn}, Server};
|
||||
|
||||
#[cfg(not(target_os = "unknown"))]
|
||||
async fn api_response(req: Request<Body>) -> Result<Response<Body>, Error> {
|
||||
match req.uri().path() {
|
||||
"/search" => {
|
||||
map_request_to_response(req, |target: Target| {
|
||||
// Filter and return metrics relating to the target
|
||||
DATABASE.read()
|
||||
.keys_starting_with(&target.target)
|
||||
.collect::<Vec<_>>()
|
||||
}).await
|
||||
},
|
||||
"/query" => {
|
||||
map_request_to_response(req, |query: Query| {
|
||||
let metrics = DATABASE.read();
|
||||
|
||||
let Query {
|
||||
range: Range { from, to },
|
||||
max_datapoints, ..
|
||||
} = query;
|
||||
|
||||
// Return timeseries data related to the specified metrics
|
||||
query.targets.iter()
|
||||
.map(|target| {
|
||||
let datapoints = metrics.datapoints_between(&target.target, from, to, max_datapoints)
|
||||
.unwrap_or_else(Vec::new);
|
||||
|
||||
TimeseriesData {
|
||||
target: target.target.clone(), datapoints
|
||||
}
|
||||
})
|
||||
.collect::<Vec<_>>()
|
||||
}).await
|
||||
},
|
||||
_ => Ok(Response::new(Body::empty())),
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(not(target_os = "unknown"))]
|
||||
async fn map_request_to_response<Req, Res, T>(req: Request<Body>, transformation: T) -> Result<Response<Body>, Error>
|
||||
where
|
||||
Req: DeserializeOwned,
|
||||
Res: Serialize,
|
||||
T: Fn(Req) -> Res + Send + Sync + 'static
|
||||
{
|
||||
let body = req.into_body()
|
||||
.map_ok(|bytes| bytes.to_vec())
|
||||
.try_concat()
|
||||
.await
|
||||
.map_err(Error::Hyper)?;
|
||||
|
||||
let req = serde_json::from_slice(body.as_ref()).map_err(Error::Serde)?;
|
||||
let res = transformation(req);
|
||||
let string = serde_json::to_string(&res).map_err(Error::Serde)?;
|
||||
|
||||
Response::builder()
|
||||
.header(header::CONTENT_TYPE, "application/json")
|
||||
.body(Body::from(string))
|
||||
.map_err(Error::Http)
|
||||
}
|
||||
|
||||
/// Given that we're not using hyper's tokio feature, we need to define out own executor.
|
||||
#[derive(Clone)]
|
||||
pub struct Executor;
|
||||
|
||||
#[cfg(not(target_os = "unknown"))]
|
||||
impl<T> hyper::rt::Executor<T> for Executor
|
||||
where
|
||||
T: Future + Send + 'static,
|
||||
T::Output: Send + 'static,
|
||||
{
|
||||
fn execute(&self, future: T) {
|
||||
async_std::task::spawn(future);
|
||||
}
|
||||
}
|
||||
|
||||
/// Start the data source server.
|
||||
#[cfg(not(target_os = "unknown"))]
|
||||
pub async fn run_server(mut address: std::net::SocketAddr) -> Result<(), Error> {
|
||||
use async_std::{net, io};
|
||||
use crate::networking::Incoming;
|
||||
|
||||
let listener = loop {
|
||||
let listener = net::TcpListener::bind(&address).await;
|
||||
match listener {
|
||||
Ok(listener) => {
|
||||
log::info!("Grafana data source server started at {}", address);
|
||||
break listener
|
||||
},
|
||||
Err(err) => match err.kind() {
|
||||
io::ErrorKind::AddrInUse | io::ErrorKind::PermissionDenied if address.port() != 0 => {
|
||||
log::warn!(
|
||||
"Unable to bind grafana data source server to {}. Trying random port.",
|
||||
address
|
||||
);
|
||||
address.set_port(0);
|
||||
continue;
|
||||
},
|
||||
_ => return Err(err.into()),
|
||||
}
|
||||
}
|
||||
};
|
||||
|
||||
let service = make_service_fn(|_| {
|
||||
async {
|
||||
Ok::<_, Error>(service_fn(api_response))
|
||||
}
|
||||
});
|
||||
|
||||
let server = Server::builder(Incoming(listener.incoming()))
|
||||
.executor(Executor)
|
||||
.serve(service)
|
||||
.boxed();
|
||||
|
||||
let every = std::time::Duration::from_secs(24 * 3600);
|
||||
let clean = clean_up(every, Duration::weeks(1))
|
||||
.boxed();
|
||||
|
||||
let result = match select(server, clean).await {
|
||||
Either::Left((result, _)) => result.map_err(Into::into),
|
||||
Either::Right((result, _)) => result
|
||||
};
|
||||
|
||||
result
|
||||
}
|
||||
|
||||
#[cfg(target_os = "unknown")]
|
||||
pub async fn run_server(_: std::net::SocketAddr) -> Result<(), Error> {
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Periodically remove old metrics.
|
||||
async fn clean_up(every: std::time::Duration, before: Duration) -> Result<(), Error> {
|
||||
loop {
|
||||
Delay::new(every).await;
|
||||
|
||||
let oldest_allowed = (Utc::now() - before).timestamp_millis();
|
||||
DATABASE.write().truncate(oldest_allowed)?;
|
||||
}
|
||||
}
|
||||
@@ -1,50 +0,0 @@
|
||||
// Copyright 2019-2020 Parity Technologies (UK) Ltd.
|
||||
// This file is part of Substrate.
|
||||
|
||||
// Substrate is free software: you can redistribute it and/or modify
|
||||
// it under the terms of the GNU General Public License as published by
|
||||
// the Free Software Foundation, either version 3 of the License, or
|
||||
// (at your option) any later version.
|
||||
|
||||
// Substrate is distributed in the hope that it will be useful,
|
||||
// but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||
// GNU General Public License for more details.
|
||||
|
||||
// You should have received a copy of the GNU General Public License
|
||||
// along with Substrate. If not, see <http://www.gnu.org/licenses/>.
|
||||
|
||||
use serde::{Serialize, Deserialize};
|
||||
|
||||
#[derive(Serialize, Deserialize)]
|
||||
pub struct Target {
|
||||
pub target: String,
|
||||
}
|
||||
|
||||
#[derive(Serialize, Deserialize)]
|
||||
pub struct Query {
|
||||
#[serde(rename = "maxDataPoints")]
|
||||
pub max_datapoints: usize,
|
||||
pub targets: Vec<Target>,
|
||||
pub range: Range,
|
||||
}
|
||||
|
||||
#[derive(Serialize, Deserialize)]
|
||||
pub struct Range {
|
||||
#[serde(deserialize_with = "date_to_timestamp_ms")]
|
||||
pub from: i64,
|
||||
#[serde(deserialize_with = "date_to_timestamp_ms")]
|
||||
pub to: i64,
|
||||
}
|
||||
|
||||
// Deserialize a timestamp via a `DateTime<Utc>`
|
||||
fn date_to_timestamp_ms<'de, D: serde::Deserializer<'de>>(timestamp: D) -> Result<i64, D::Error> {
|
||||
Deserialize::deserialize(timestamp)
|
||||
.map(|date: chrono::DateTime<chrono::Utc>| date.timestamp_millis())
|
||||
}
|
||||
|
||||
#[derive(Serialize, Deserialize)]
|
||||
pub struct TimeseriesData {
|
||||
pub target: String,
|
||||
pub datapoints: Vec<(f32, i64)>
|
||||
}
|
||||
@@ -1,13 +0,0 @@
|
||||
[package]
|
||||
description = "Grafana data source server test"
|
||||
name = "grafana-data-source-test"
|
||||
version = "2.0.0"
|
||||
license = "GPL-3.0"
|
||||
authors = ["Parity Technologies <admin@parity.io>"]
|
||||
edition = "2018"
|
||||
|
||||
[dependencies]
|
||||
grafana-data-source = { version = "0.8", path = ".." }
|
||||
futures = "0.3"
|
||||
futures-timer = "3.0.1"
|
||||
rand = "0.7"
|
||||
@@ -1,44 +0,0 @@
|
||||
// Copyright 2019-2020 Parity Technologies (UK) Ltd.
|
||||
// This file is part of Substrate.
|
||||
|
||||
// Substrate is free software: you can redistribute it and/or modify
|
||||
// it under the terms of the GNU General Public License as published by
|
||||
// the Free Software Foundation, either version 3 of the License, or
|
||||
// (at your option) any later version.
|
||||
|
||||
// Substrate is distributed in the hope that it will be useful,
|
||||
// but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||
// GNU General Public License for more details.
|
||||
|
||||
// You should have received a copy of the GNU General Public License
|
||||
// along with Substrate. If not, see <http://www.gnu.org/licenses/>.
|
||||
|
||||
use grafana_data_source::{run_server, record_metrics};
|
||||
use std::time::Duration;
|
||||
use rand::Rng;
|
||||
use futures::{future::join, executor};
|
||||
|
||||
async fn randomness() {
|
||||
loop {
|
||||
futures_timer::Delay::new(Duration::from_secs(1)).await;
|
||||
|
||||
let random = rand::thread_rng().gen_range(0.0, 1000.0);
|
||||
|
||||
let result = record_metrics!(
|
||||
"random data" => random,
|
||||
"random^2" => random * random,
|
||||
);
|
||||
|
||||
if let Err(error) = result {
|
||||
eprintln!("{}", error);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
fn main() {
|
||||
executor::block_on(join(
|
||||
run_server("127.0.0.1:9955".parse().unwrap()),
|
||||
randomness()
|
||||
)).0.unwrap();
|
||||
}
|
||||
+3
-8
@@ -1,6 +1,6 @@
|
||||
[package]
|
||||
description = "Grafana data source server"
|
||||
name = "grafana-data-source"
|
||||
description = "Prometheus exporter server"
|
||||
name = "prometheus-exporter"
|
||||
version = "0.8.0"
|
||||
license = "GPL-3.0"
|
||||
authors = ["Parity Technologies <admin@parity.io>"]
|
||||
@@ -8,13 +8,8 @@ edition = "2018"
|
||||
|
||||
[dependencies]
|
||||
log = "0.4.8"
|
||||
prometheus = "0.7"
|
||||
futures-util = { version = "0.3.1", default-features = false, features = ["io"] }
|
||||
serde_json = "1"
|
||||
serde = { version = "1", features = ["derive"] }
|
||||
chrono = { version = "0.4", features = ["serde"] }
|
||||
lazy_static = "1.4"
|
||||
parking_lot = "0.10.0"
|
||||
futures-timer = "3.0.1"
|
||||
derive_more = "0.99"
|
||||
|
||||
[target.'cfg(not(target_os = "unknown"))'.dependencies]
|
||||
@@ -0,0 +1,16 @@
|
||||
# Substrate Prometheus Exporter
|
||||
|
||||
## Introduction
|
||||
|
||||
[Prometheus](https://prometheus.io/) is one of the most widely used monitoring tools for managing highly available services supported by [Cloud Native Computing Foundation](https://www.cncf.io/). By providing Prometheus metrics in Substrate, node operators can easily adopt widely used display/alert tools such
|
||||
as [Grafana](https://grafana.com/) and [Alertmanager](https://prometheus.io/docs/alerting/alertmanager/). Easy access to such monitoring tools will benefit parachain developers/operators and validators to have much higher availability of their services.
|
||||
|
||||
Metrics will be served under `/metrics` on TCP port 9615 by default.
|
||||
|
||||
## Quick Start
|
||||
|
||||
1. From the root of the repository start Substrate `cargo run --release`.
|
||||
|
||||
2. In another terminal run `curl localhost:9615/metrics` to retrieve the metrics.
|
||||
|
||||
To learn how to configure Prometheus see the Prometheus [Getting Started](https://prometheus.io/docs/prometheus/latest/getting_started/) guide.
|
||||
@@ -0,0 +1,144 @@
|
||||
// Copyright 2019 Parity Technologies (UK) Ltd.
|
||||
// This file is part of Substrate.
|
||||
|
||||
// Substrate is free software: you can redistribute it and/or modify
|
||||
// it under the terms of the GNU General Public License as published by
|
||||
// the Free Software Foundation, either version 3 of the License, or
|
||||
// (at your option) any later version.
|
||||
|
||||
// Substrate is distributed in the hope that it will be useful,
|
||||
// but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||
// GNU General Public License for more details.
|
||||
|
||||
// You should have received a copy of the GNU General Public License
|
||||
// along with Substrate. If not, see <http://www.gnu.org/licenses/>.
|
||||
|
||||
use futures_util::{FutureExt, future::Future};
|
||||
pub use prometheus::{
|
||||
Registry, Error as PrometheusError, Opts,
|
||||
core::{
|
||||
GenericGauge as Gauge, GenericCounter as Counter,
|
||||
GenericGaugeVec as GaugeVec, GenericCounterVec as CounterVec,
|
||||
AtomicF64 as F64, AtomicI64 as I64, AtomicU64 as U64,
|
||||
}
|
||||
};
|
||||
use prometheus::{Encoder, TextEncoder, core::Collector};
|
||||
use std::net::SocketAddr;
|
||||
|
||||
#[cfg(not(target_os = "unknown"))]
|
||||
mod networking;
|
||||
|
||||
#[cfg(target_os = "unknown")]
|
||||
pub use unknown_os::init_prometheus;
|
||||
#[cfg(not(target_os = "unknown"))]
|
||||
pub use known_os::init_prometheus;
|
||||
|
||||
pub fn register<T: Clone + Collector + 'static>(metric: T, registry: &Registry) -> Result<T, PrometheusError> {
|
||||
registry.register(Box::new(metric.clone()))?;
|
||||
Ok(metric)
|
||||
}
|
||||
|
||||
// On WASM `init_prometheus` becomes a no-op.
|
||||
#[cfg(target_os = "unknown")]
|
||||
mod unknown_os {
|
||||
use super::*;
|
||||
|
||||
pub enum Error {}
|
||||
|
||||
pub async fn init_prometheus(_: SocketAddr, _registry: Registry) -> Result<(), Error> {
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(not(target_os = "unknown"))]
|
||||
mod known_os {
|
||||
use super::*;
|
||||
use hyper::http::StatusCode;
|
||||
use hyper::{Server, Body, Request, Response, service::{service_fn, make_service_fn}};
|
||||
|
||||
#[derive(Debug, derive_more::Display, derive_more::From)]
|
||||
pub enum Error {
|
||||
/// Hyper internal error.
|
||||
Hyper(hyper::Error),
|
||||
/// Http request error.
|
||||
Http(hyper::http::Error),
|
||||
/// i/o error.
|
||||
Io(std::io::Error),
|
||||
#[display(fmt = "Prometheus exporter port {} already in use.", _0)]
|
||||
PortInUse(SocketAddr)
|
||||
}
|
||||
|
||||
impl std::error::Error for Error {
|
||||
fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
|
||||
match self {
|
||||
Error::Hyper(error) => Some(error),
|
||||
Error::Http(error) => Some(error),
|
||||
Error::Io(error) => Some(error),
|
||||
Error::PortInUse(_) => None
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
async fn request_metrics(req: Request<Body>, registry: Registry) -> Result<Response<Body>, Error> {
|
||||
if req.uri().path() == "/metrics" {
|
||||
let metric_families = registry.gather();
|
||||
let mut buffer = vec![];
|
||||
let encoder = TextEncoder::new();
|
||||
encoder.encode(&metric_families, &mut buffer).unwrap();
|
||||
|
||||
Response::builder().status(StatusCode::OK)
|
||||
.header("Content-Type", encoder.format_type())
|
||||
.body(Body::from(buffer))
|
||||
.map_err(Error::Http)
|
||||
} else {
|
||||
Response::builder().status(StatusCode::NOT_FOUND)
|
||||
.body(Body::from("Not found."))
|
||||
.map_err(Error::Http)
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
#[derive(Clone)]
|
||||
pub struct Executor;
|
||||
|
||||
impl<T> hyper::rt::Executor<T> for Executor
|
||||
where
|
||||
T: Future + Send + 'static,
|
||||
T::Output: Send + 'static,
|
||||
{
|
||||
fn execute(&self, future: T) {
|
||||
async_std::task::spawn(future);
|
||||
}
|
||||
}
|
||||
|
||||
/// Initializes the metrics context, and starts an HTTP server
|
||||
/// to serve metrics.
|
||||
pub async fn init_prometheus(prometheus_addr: SocketAddr, registry: Registry) -> Result<(), Error>{
|
||||
use networking::Incoming;
|
||||
let listener = async_std::net::TcpListener::bind(&prometheus_addr)
|
||||
.await
|
||||
.map_err(|_| Error::PortInUse(prometheus_addr))?;
|
||||
|
||||
log::info!("Prometheus server started at {}", prometheus_addr);
|
||||
|
||||
let service = make_service_fn(move |_| {
|
||||
let registry = registry.clone();
|
||||
|
||||
async move {
|
||||
Ok::<_, hyper::Error>(service_fn(move |req: Request<Body>| {
|
||||
request_metrics(req, registry.clone())
|
||||
}))
|
||||
}
|
||||
});
|
||||
|
||||
let server = Server::builder(Incoming(listener.incoming()))
|
||||
.executor(Executor)
|
||||
.serve(service)
|
||||
.boxed();
|
||||
|
||||
let result = server.await.map_err(Into::into);
|
||||
|
||||
result
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user