WIP Introduce xcp-metrics daemon

xcp-metrics exposes the same kind of metrics that xcp-rrdd does, but
using the OpenMetrics standard.

It uses an reworked version of the v2 protocol currently in use by xcp-rrdd,
proposed as v3 protocol in xapi-project/xapi-project.github.io#278

Signed-off-by: Teddy Astie <[email protected]>
Reviewed-by: Yann Dirson <[email protected]>

Cargo.toml

@@ -3,10 +3,12 @@ package.version = "0.1.0"
 package.repository = "https://github.com/xcp-ng/xcp-metrics"
 package.categories = ["virtualization"]
 members = [
+  "xcp-metrics",
   "xcp-metrics-common",
   "xapi-rs",
   "plugins/xcp-metrics-plugin-common",
   "plugins/xcp-metrics-plugin-squeezed",
+  "xcp-metrics-tools",
 ]
 [profile.release]
 lto = true

xcp-metrics-tools/Cargo.toml

@@ -0,0 +1,25 @@
+[package]
+name = "xcp-metrics-tools"
+description = "Various xcp-metrics utilities"
+version.workspace = true
+license = "AGPL-3.0-only"
+edition = "2021"
+
+# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
+
+[dependencies]
+xcp-metrics-common = { path = "../xcp-metrics-common" }
+xapi = { path = "../xapi-rs" }
+
+anyhow = "1.0"
+serde_json = "1.0"
+
+tokio = { version = "1", features = ["full"] }
+
+[dependencies.clap]
+version = "4.3"
+features = ["derive"]
+
+[dependencies.hyper]
+version = "0.14"
+features = ["full"]

xcp-metrics-tools/README.md

@@ -0,0 +1,17 @@
+# xcp-metrics tools
+
+Set of various xcp-metrics utilities.
+
+## Project binaries
+
+### xcp-metrics-dump
+
+Utility to dump plugin protocol v2 or protocol v3 files.
+
+### xcp-metrics-get-metrics
+
+Tool that fetches current OpenMetrics from xcp-metrics daemon using either the protobuf or text format.
+
+### xcp-metrics-openmetrics-proxy
+
+Small program that redirects HTTP requests into OpenMetrics RPC route, meant to be used with Prometheus or similar projects.

xcp-metrics-tools/src/bin/xcp-metrics-dump.rs

@@ -0,0 +1,73 @@
+use anyhow::Result;
+use std::{
+    fs,
+    io::{Read, Seek},
+};
+use xcp_metrics_common::{
+    self, protocol_v3,
+    rrdd::protocol_v2::{self, RrddMessageHeader, RrddMetadata, RrddMetadataRaw},
+};
+
+fn read_protocol_v2(mut file: fs::File) -> Result<(), anyhow::Error> {
+    let header = RrddMessageHeader::parse_from(&mut file)?;
+
+    println!("{:#?}", &header);
+
+    let mut buffer = vec![0u8; header.metadata_length as usize];
+    file.read_exact(&mut buffer)?;
+
+    let metadata_raw: RrddMetadataRaw = serde_json::from_slice(&buffer)?;
+    let metadata = RrddMetadata::try_from(metadata_raw);
+
+    println!("{metadata:#?}");
+
+    Ok(())
+}
+
+fn read_protocol_v3(mut file: fs::File) -> Result<(), anyhow::Error> {
+    match protocol_v3::parse_v3(&mut file) {
+        Ok((header, metrics)) => {
+            println!("{header:#?}");
+            println!("{metrics:#?}");
+        }
+        Err(e) => {
+            println!("Read failure ({e}), try reading header, skipping crc checks.");
+
+            file.rewind()?;
+            println!(
+                "{:#?}",
+                protocol_v3::ProtocolV3Header::parse_from(&mut file)
+            );
+        }
+    }
+
+    Ok(())
+}
+
+fn main() -> Result<()> {
+    let args: Vec<String> = std::env::args().collect();
+
+    if let Some(path) = args.get(1) {
+        println!("Trying to read message header...");
+        let mut file = fs::File::open(path)?;
+
+        let mut file_header = [0u8; 12];
+        file.read_exact(&mut file_header)?;
+
+        file.rewind()?;
+
+        if file_header == *protocol_v3::PROTOCOL_V3_HEADER {
+            println!("Detected protocol v3");
+            read_protocol_v3(file)?;
+        } else if file_header[..11] == *protocol_v2::PROTOCOL_V2_HEADER {
+            println!("Detected protocol v2");
+            read_protocol_v2(file)?;
+        } else {
+            println!("Unknown file header");
+        }
+    } else {
+        println!("Usage: xcp-metrics-dump /dev/shm/metrics/<file>");
+    }
+
+    Ok(())
+}

xcp-metrics-tools/src/bin/xcp-metrics-get-metrics.rs

@@ -0,0 +1,77 @@
+use std::path::PathBuf;
+
+use clap::Parser;
+use tokio::io::{stdout, AsyncWriteExt};
+use xapi::{
+    hyper::{self, body, Body},
+    hyperlocal,
+    rpc::{
+        message::RpcKind, methods::OpenMetricsMethod, write_method_jsonrpc, write_method_xmlrpc,
+    },
+};
+
+/// Tool to get metrics from xcp-metrics in OpenMetrics format.
+#[derive(Parser, Debug)]
+#[command(author, version, about, long_about = None)]
+struct Args {
+    /// Path to the xcp-metrics daemon socket to fetch metrics from.
+    #[arg(short, long)]
+    daemon_path: Option<PathBuf>,
+
+    /// RPC format to use
+    #[arg(long, default_value_t = RpcKind::JsonRpc)]
+    rpc_format: RpcKind,
+
+    /// Whether to use protocol buffers binary format.
+    #[arg(short, long, default_value_t = false)]
+    binary: bool,
+}
+
+#[tokio::main]
+async fn main() {
+    let args = Args::parse();
+    let daemon_path = args
+        .daemon_path
+        .unwrap_or_else(|| xapi::get_module_path("xcp-metrics"));
+
+    let module_uri = hyperlocal::Uri::new(daemon_path, "/");
+
+    let mut rpc_buffer = vec![];
+    let method = OpenMetricsMethod {
+        protobuf: args.binary,
+    };
+
+    match args.rpc_format {
+        RpcKind::JsonRpc => write_method_jsonrpc(&mut rpc_buffer, &method).unwrap(),
+        RpcKind::XmlRpc => write_method_xmlrpc(&mut rpc_buffer, &method).unwrap(),
+    };
+
+    let content_type = match args.rpc_format {
+        RpcKind::JsonRpc => "application/json-rpc",
+        RpcKind::XmlRpc => "application/xml",
+    };
+
+    eprintln!("Sent: {}", String::from_utf8_lossy(&rpc_buffer));
+
+    let request = hyper::Request::builder()
+        .uri(hyper::Uri::from(module_uri))
+        .method("POST")
+        .header("User-agent", "xcp-metrics-get-metrics")
+        .header("content-length", rpc_buffer.len())
+        .header("content-type", content_type)
+        .header("host", "localhost")
+        .body(Body::from(rpc_buffer))
+        .unwrap();
+
+    let response = hyper::Client::builder()
+        .build(hyperlocal::UnixConnector)
+        .request(request)
+        .await;
+
+    eprintln!("{response:#?}");
+
+    let response = response.unwrap();
+    let data = body::to_bytes(response.into_body()).await.unwrap();
+
+    stdout().write_all(&data).await.unwrap();
+}

xcp-metrics-tools/src/bin/xcp-metrics-openmetrics-proxy.rs

@@ -0,0 +1,63 @@
+use std::{
+    net::{IpAddr, Ipv4Addr, SocketAddr},
+    path::{Path, PathBuf},
+};
+
+use clap::{command, Parser};
+use hyper::{
+    server::{conn::AddrStream, Server},
+    service::{make_service_fn, service_fn},
+    Body, Request, Response,
+};
+
+use xapi::rpc::methods::OpenMetricsMethod;
+
+/// OpenMetrics http proxy, used to provide metrics for collectors such as Prometheus.
+#[derive(Clone, Parser, Debug)]
+#[command(author, version, about, long_about = None)]
+struct Args {
+    /// Adress to bind the HTTP server to.
+    #[arg(short, long, default_value_t = SocketAddr::new(IpAddr::V4(Ipv4Addr::UNSPECIFIED), 8080))]
+    addr: SocketAddr,
+
+    /// Path to the xcp-metrics daemon socket to fetch metrics from.
+    #[arg(short, long)]
+    daemon_path: Option<PathBuf>,
+}
+
+async fn redirect_openmetrics(
+    request: Request<Body>,
+    daemon_path: &Path,
+) -> anyhow::Result<Response<Body>> {
+    xapi::send_jsonrpc_to(
+        daemon_path,
+        "POST",
+        &OpenMetricsMethod::default(),
+        "xcp-metrics-openmetrics-proxy",
+    )
+    .await
+}
+
+#[tokio::main]
+async fn main() {
+    let args = Args::parse();
+    let daemon_path = args
+        .daemon_path
+        .unwrap_or_else(|| xapi::get_module_path("xcp-metrics"));
+
+    let service_fn = make_service_fn(|addr: &AddrStream| {
+        println!("Handling request {:?}", addr);
+        let daemon_path = daemon_path.clone();
+
+        async {
+            anyhow::Ok(service_fn(move |request| {
+                let daemon_path = daemon_path.clone();
+                async move { redirect_openmetrics(request, &daemon_path).await }
+            }))
+        }
+    });
+
+    let server = Server::bind(&args.addr).serve(service_fn);
+
+    server.await.expect("Proxy server failure");
+}

xcp-metrics/Cargo.toml

@@ -0,0 +1,40 @@
+[package]
+name = "xcp-metrics"
+description = "Main xcp-metrics daemon"
+version.workspace = true
+license = "AGPL-3.0-only"
+edition = "2021"
+
+# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
+
+[dependencies]
+xcp-metrics-common = { path = "../xcp-metrics-common" }
+xapi = { path = "../xapi-rs" }
+
+serde_json = "1.0"
+anyhow = "1.0"
+futures = "0.3"
+dashmap = "5.4.0"
+tracing = "0.1"
+tracing-subscriber = "0.3"
+maplit = "1.0.2"
+
+[dependencies.tokio]
+version = "1"
+features = ["full"]
+
+[dependencies.serde]
+version = "1.0"
+features = ["std", "derive"]
+
+[dependencies.uuid]
+version = "1.4"
+features = ["std", "serde", "v4", "fast-rng"]
+
+[dependencies.sendfd]
+version = "0.4.3"
+features = ["tokio"]
+
+[dependencies.clap]
+version = "4.3"
+features = ["derive"]

xcp-metrics/README.md

@@ -0,0 +1,26 @@
+# xcp-metrics main daemon
+
+This is the main xcp-metrics daemon that uses a RPC interface similar to xcp-rrdd (and what some other XAPI project uses).
+In addition to support XML-RPC, it also supports JSON-RPC.
+
+## Main modules
+
+### forwarded
+
+Forwarded implementation and routes (e.g `rrd_updates`) that manages the forwarded socket (e.g `xcp-rrdd.forwarded`).
+
+### hub
+
+Small module that aggregate metrics.
+
+### providers
+
+Metrics providers implementations (e.g protocol v2 and v3) that pushes metrics to hub.
+
+### publishers
+
+Modules that pulls metrics from hub and distribute them (using RPC, forwarded route or something else).
+
+### rpc
+
+RPC server implementation and routes.