Fix folder names in client and test (#4360)

* client/rpc/api -> client/rpc-api

* client/util/wasm-builder-runner -> utils/wasm-builder-runner

* client/grafana-data-source -> utils/grafana-data-source

* test/utils -> test-utils

* fix moved path

* Update Cargo.lock

* Update Cargo.lock

substrate/utils/grafana-data-source/src/database.rs

@@ -0,0 +1,170 @@
+// 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 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(&timestamp, |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();
+	let start = now_millis();
+
+	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"]);
+
+	assert_eq!(
+		database.datapoints_between("test", start - 1000, start + 1000, 4),
+		Some(vec![(1.0, start), (2.5, start), (2.0, start)])
+	);
+
+	assert_eq!(
+		database.datapoints_between("test", start - 1000, start + 1000, 3),
+		Some(vec![(1.0, start), (2.5, start), (2.0, start)])
+	);
+
+	assert_eq!(
+		database.datapoints_between("test", start - 1000, start + 1000, 2),
+		Some(vec![(1.0, start), (2.0, start)])
+	);
+}

substrate/utils/grafana-data-source/src/lib.rs

@@ -0,0 +1,91 @@
+// 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
+
+#![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(hyper::Error),
+	Serde(serde_json::Error),
+	Http(hyper::http::Error),
+	Timestamp(TryFromIntError),
+	Io(std::io::Error)
+}
+
+impl std::error::Error for Error {
+	fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
+		match self {
+			Error::Hyper(error) => Some(error),
+			Error::Serde(error) => Some(error),
+			Error::Http(error) => Some(error),
+			Error::Timestamp(error) => Some(error),
+			Error::Io(error) => Some(error)
+		}
+	}
+}

substrate/utils/grafana-data-source/src/networking.rs

@@ -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)
+	}
+}

substrate/utils/grafana-data-source/src/server.rs

@@ -0,0 +1,162 @@
+// 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::{DATABASE, Error, types::{Target, Query, TimeseriesData, Range}};
+
+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())),
+	}
+}
+
+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(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;
+				},
+				_ => Err(err)?,
+			}
+		}
+	};
+
+	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)?;
+	}
+}

substrate/utils/grafana-data-source/src/types.rs

@@ -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)>
+}