Grafana integration (#3913)

* Very WIP * record_metrics macro works * Integrate into service * Licenses and documentation * Remove unused Debugs, make respond function clearer * Conform to line widths, fix service test * Switch to storing the timestamps as millis instead * Update core/grafana-data-source/src/lib.rs Co-Authored-By: Kian Paimani <5588131+kianenigma@users.noreply.github.com> * Transform timestamps to i64 in serialization * Fix license date * Binary sort to find selection range for metrics * Obey maxDataPoints * Run a cleaning future * Newlines at EOF * Update core/service/Cargo.toml Co-Authored-By: Pierre Krieger <pierre.krieger1708@gmail.com> * Update core/grafana-data-source/src/lib.rs Co-Authored-By: Pierre Krieger <pierre.krieger1708@gmail.com> * Fix indentation * Improve select_points * Made test more accurate * Inprogress * Use the same futures version as hyper for now * Error handling * Remove dependence on hyper's tokio feature * Added target_os flag * Update Cargo.toml Co-Authored-By: Pierre Krieger <pierre.krieger1708@gmail.com> * Simplify example * Remove compat wildcard * Updated lock file * Fix indentation 😉
2026-04-27 02:17:58 +00:00 · 2019-11-22 15:49:49 +01:00
parent 458dba5ce5
commit 145efab68f
16 changed files with 805 additions and 1 deletions
@@ -0,0 +1,58 @@
+// 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/>.
+
+//! [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
+
+use lazy_static::lazy_static;
+use std::collections::HashMap;
+use parking_lot::RwLock;
+
+mod types;
+mod server;
+mod util;
+#[cfg(not(target_os = "unknown"))]
+mod networking;
+
+pub use server::run_server;
+pub use util::now_millis;
+
+type Metrics = HashMap<&'static str, Vec<(f32, i64)>>;
+
+lazy_static! {
+	/// The `RwLock` wrapping the metrics. Not intended to be used directly.
+	#[doc(hidden)]
+	pub static ref METRICS: RwLock<Metrics> = RwLock::new(Metrics::new());
+}
+
+/// Write metrics to `METRICS`.
+#[macro_export]
+macro_rules! record_metrics(
+	($($key:expr => $value:expr),*) => {
+		use $crate::{METRICS, now_millis};
+		let mut metrics = METRICS.write();
+		let now = now_millis();
+		$(
+			metrics.entry($key).or_insert_with(Vec::new).push(($value as f32, now));
+		)*
+	}
+);
@@ -0,0 +1,66 @@
+// 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 async_std::pin::Pin;
+use std::task::{Poll, Context};
+use futures_util::{stream::Stream, io::{AsyncRead, AsyncWrite}};
+
+pub struct Incoming<'a>(pub async_std::net::Incoming<'a>);
+
+impl hyper::server::accept::Accept for Incoming<'_> {
+	type Conn = TcpStream;
+	type Error = async_std::io::Error;
+
+	fn poll_accept(self: Pin<&mut Self>, cx: &mut Context) -> Poll<Option<Result<Self::Conn, Self::Error>>> {
+		Pin::new(&mut Pin::into_inner(self).0)
+			.poll_next(cx)
+			.map(|opt| opt.map(|res| res.map(TcpStream)))
+	}
+}
+
+pub struct TcpStream(pub async_std::net::TcpStream);
+
+impl tokio_io::AsyncRead for TcpStream {
+	fn poll_read(
+		self: Pin<&mut Self>,
+		cx: &mut Context,
+		buf: &mut [u8]
+	) -> Poll<Result<usize, std::io::Error>> {
+		Pin::new(&mut Pin::into_inner(self).0)
+			.poll_read(cx, buf)
+	}
+}
+
+impl tokio_io::AsyncWrite for TcpStream {
+	fn poll_write(
+		self: Pin<&mut Self>,
+		cx: &mut Context,
+		buf: &[u8]
+	) -> Poll<Result<usize, std::io::Error>> {
+		Pin::new(&mut Pin::into_inner(self).0)
+			.poll_write(cx, buf)
+	}
+
+	fn poll_flush(self: Pin<&mut Self>, cx: &mut Context) -> Poll<Result<(), std::io::Error>> {
+		Pin::new(&mut Pin::into_inner(self).0)
+			.poll_flush(cx)
+	}
+
+	fn poll_shutdown(self: Pin<&mut Self>, cx: &mut Context) -> Poll<Result<(), std::io::Error>> {
+		Pin::new(&mut Pin::into_inner(self).0)
+			.poll_close(cx)
+	}
+}
@@ -0,0 +1,165 @@
+// 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 serde::{Serialize, de::DeserializeOwned};
+use hyper::{Body, Request, Response, header, service::{service_fn, make_service_fn}, Server};
+use chrono::{Duration, Utc};
+use futures_util::{FutureExt, future::{Future, select, Either}};
+use futures_timer::Delay;
+use crate::{METRICS, util, types::{Target, Query, TimeseriesData}};
+
+#[derive(Debug, derive_more::Display)]
+enum Error {
+	Hyper(hyper::Error),
+	Serde(serde_json::Error),
+	Http(hyper::http::Error)
+}
+
+impl std::error::Error for Error {}
+
+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
+				METRICS.read()
+					.keys()
+					.filter(|key| key.starts_with(&target.target))
+					.cloned()
+					.collect::<Vec<_>>()
+			}).await
+		},
+		"/query" => {
+			map_request_to_response(req, |query: Query| {
+				let metrics = METRICS.read();
+
+				// Return timeseries data related to the specified metrics
+				query.targets.iter()
+					.map(|target| {
+						let datapoints = metrics.get(target.target.as_str())
+							.map(|metric| {
+								let from = util::find_index(&metric, query.range.from);
+								let to = util::find_index(&metric, query.range.to);
+
+								// Avoid returning more than `max_datapoints` (mostly to stop
+								// the web browser from having to do a ton of work)
+								util::select_points(&metric[from .. to], query.max_datapoints)
+							})
+							.unwrap_or_else(Vec::new);
+
+						TimeseriesData {
+							target: target.target.clone(), datapoints
+						}
+					})
+					.collect::<Vec<_>>()
+			}).await
+		},
+		_ => Ok(Response::new(Body::empty())),
+	}
+}
+
+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
+{
+	use futures_util_alpha::TryStreamExt;
+
+	let body = req.into_body()
+		.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> tokio_executor::TypedExecutor<T> for Executor
+	where
+		T: Future + Send + 'static,
+		T::Output: Send + 'static,
+{
+	fn spawn(&mut self, future: T) -> Result<(), tokio_executor::SpawnError> {
+		async_std::task::spawn(future);
+		Ok(())
+	}
+}
+
+/// Start the data source server.
+#[cfg(not(target_os = "unknown"))]
+pub async fn run_server(address: std::net::SocketAddr) -> Result<(), hyper::Error> {
+	use crate::networking::Incoming;
+
+	let listener = async_std::net::TcpListener::bind(&address).await.unwrap();
+
+	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 clean = clean_up(Duration::days(1), Duration::weeks(1))
+		.boxed();
+
+	let result = match select(server, clean).await {
+		Either::Left((result, _)) => result,
+		Either::Right(_) => Ok(())
+	};
+
+	result
+}
+
+#[cfg(target_os = "unknown")]
+pub async fn run_server(_: std::net::SocketAddr) -> Result<(), hyper::Error> {
+	Ok(())
+}
+
+/// Periodically remove old metrics.
+async fn clean_up(every: Duration, before: Duration) {
+	loop {
+		Delay::new(every.to_std().unwrap()).await;
+
+		let oldest_allowed = (Utc::now() - before).timestamp_millis();
+
+		let mut metrics = METRICS.write();
+
+		for metric in metrics.values_mut() {
+			// Find the index of the oldest allowed timestamp and cut out all those before it.
+			let index = util::find_index(&metric, oldest_allowed);
+
+			if index > 0 {
+				*metric = metric[index..].to_vec();
+			}
+		}
+	}
+}
@@ -0,0 +1,50 @@
+// 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 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)>
+}
@@ -0,0 +1,52 @@
+// 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/>.
+
+/// Get the current unix timestamp in milliseconds.
+pub fn now_millis() -> i64 {
+	chrono::Utc::now().timestamp_millis()
+}
+
+// find the index of a timestamp
+pub fn find_index(slice: &[(f32, i64)], timestamp: i64) -> usize {
+	slice.binary_search_by_key(&timestamp, |&(_, timestamp)| timestamp)
+		.unwrap_or_else(|index| index)
+}
+
+// Evenly select up to `num_points` points from a slice
+pub fn select_points<T: Copy>(slice: &[T], num_points: usize) -> Vec<T> {
+	if num_points == 0 {
+		return Vec::new();
+	} else if num_points >= slice.len() {
+		return slice.to_owned();
+	}
+
+	(0 .. num_points - 1)
+		.map(|i| slice[i * slice.len() / (num_points - 1)])
+		.chain(slice.last().cloned())
+		.collect()
+}
+
+#[test]
+fn test_select_points() {
+	let array = [1, 2, 3, 4, 5];
+	assert_eq!(select_points(&array, 0), Vec::<u8>::new());
+	assert_eq!(select_points(&array, 1), vec![5]);
+	assert_eq!(select_points(&array, 2), vec![1, 5]);
+	assert_eq!(select_points(&array, 3), vec![1, 3, 5]);
+	assert_eq!(select_points(&array, 4), vec![1, 2, 4, 5]);
+	assert_eq!(select_points(&array, 5), vec![1, 2, 3, 4, 5]);
+	assert_eq!(select_points(&array, 6), vec![1, 2, 3, 4, 5]);
+}