Continue reorganization

* introduce xcp-metrics protocol * port and redesign xenstore plugin * rework xcp-metrics-tools Signed-off-by: Teddy Astie <[email protected]>
xcp-ng · Jan 7, 2025 · 1eade29 · 1eade29
1 parent 2a549af
commit 1eade29
Show file tree

Hide file tree

Showing 16 changed files with 378 additions and 392 deletions.
diff --git a/Cargo.toml b/Cargo.toml
@@ -6,18 +6,17 @@ package.categories = ["virtualization"]
 members = [
   "xcp-metrics",
   "xcp-metrics-common",
-  "xapi-rs",
-  "xcp-metrics-test",
-  "plugins/xcp-metrics-plugin-common",
-  "plugins/xcp-metrics-plugin-xen",
-  "plugins/xcp-metrics-plugin-xen-v2",
+  #"xapi-rs",
+  #"xcp-metrics-test",
+  #"plugins/xcp-metrics-plugin-common",
+  #"plugins/xcp-metrics-plugin-xen",
   "plugins/xcp-metrics-plugin-xenstored",
-  "plugins/xcp-metrics-plugin-squeezed",
-  "plugins/xcp-metrics-plugin-procfs",
-  "plugins/xcp-metrics-plugin-tests",
-  "plugins/xcp-metrics-plugin-bridge-v2",
+  #"plugins/xcp-metrics-plugin-squeezed",
+  #"plugins/xcp-metrics-plugin-procfs",
+  #"plugins/xcp-metrics-plugin-tests",
+  #"plugins/xcp-metrics-plugin-bridge-v2",
   "xcp-metrics-tools",
 ]
 
 [profile.release]
-lto = true
+lto = true
diff --git a/plugins/xcp-metrics-plugin-common/Cargo.toml b/plugins/xcp-metrics-plugin-common/Cargo.toml
@@ -13,24 +13,11 @@ xapi = { path = "../../xapi-rs" }
 
 anyhow = "1.0"
 tokio = "1"
-uuid = "1.4"
+uuid = "1.11"
 tracing = "0.1"
 tracing-subscriber = "0.3"
 
-dashmap = "5.5"
+dashmap = "6.1"
 futures = "0.3"
 
-[dependencies.xenstore-rs]
-version = "0.3"
-optional = true
-
-[dependencies.xenstore-rs-wip]
-git = "https://github.com/xcp-ng/xenstore-rs"
-package = "xenstore-rs"
-branch = "xs_watch"
-optional = true
-
-[features]
-default = []
-xenstore = ["dep:xenstore-rs"]
-xenstore-wip = ["xenstore", "dep:xenstore-rs-wip"]
+xenstore-rs = "0.8"
diff --git a/plugins/xcp-metrics-plugin-common/src/bridge/v3_to_v2.rs b/plugins/xcp-metrics-plugin-common/src/bridge/v3_to_v2.rs
@@ -2,7 +2,7 @@
 use std::{collections::HashMap, iter};
 
 use xcp_metrics_common::{
-    metrics::{Label, Metric, MetricFamily, MetricPoint, MetricSet, MetricValue},
+    metrics::{Label, Metric, MetricFamily, MetricValue, MetricSet, MetricValue},
     rrdd::{
         protocol_common::{DataSourceMetadata, DataSourceValue},
         protocol_v2::{indexmap::IndexMap, RrddMetadata},
@@ -14,6 +14,7 @@ use xcp_metrics_common::{
 };
 
 /// Adapter to convert protocol v3 metrics set into protocol v2 metadata and data.
+#[derive(Clone)]
 pub struct BridgeToV2 {
     model: MetricSetModel,
     latest_set: MetricSet,
@@ -24,7 +25,7 @@ pub struct BridgeToV2 {
 }
 
 /// Convert a MetricPoint into a protocol-v2 value.
-fn metric_point_to_v2(metric_point: &MetricPoint) -> DataSourceValue {
+fn metric_point_to_v2(metric_point: &MetricValue) -> DataSourceValue {
     match metric_point.value {
         MetricValue::Gauge(value) => DataSourceValue::from(value),
         MetricValue::Counter { total, .. } => DataSourceValue::from(total),
@@ -99,7 +100,7 @@ impl BridgeToV2 {
         &'a self,
         family_name: &str,
         labels: &[Label],
-    ) -> Option<&'a MetricPoint> {
+    ) -> Option<&'a MetricValue> {
         self.latest_set
             .families
             .get(family_name)

diff --git a/plugins/xcp-metrics-plugin-common/src/lib.rs b/plugins/xcp-metrics-plugin-common/src/lib.rs
@@ -3,5 +3,4 @@ pub mod plugin;
 pub mod protocol_v2;
 pub mod protocol_v3;
 
-#[cfg(feature = "xenstore")]
-pub mod xenstore;
+//pub mod xenstore;
diff --git a/plugins/xcp-metrics-plugin-common/src/plugin.rs b/plugins/xcp-metrics-plugin-common/src/plugin.rs
@@ -16,7 +16,7 @@ pub const XCP_RRDD_PATH: &str = "/var/lib/xcp/xcp-rrdd";
 /// Abstraction of a protocol v3 plugin.
 pub trait XcpPlugin {
     /// Update the state of the plugin.
-    fn update(&mut self);
+    fn update(&mut self) -> impl std::future::Future<Output = ()> + Send;
 
     // Generate a new metric set representing the current state of data.
     fn generate_metrics(&mut self) -> SimpleMetricSet;
@@ -93,7 +93,7 @@ pub async fn run_plugin_v2(mut shared: impl XcpPlugin, target_daemon_path: Optio
         tracing::debug!("Updating plugin state");
 
         // Update sources
-        shared.update();
+        shared.update().await;
 
         // Fetch and push new metrics.
         metrics = shared.generate_metrics();
@@ -130,7 +130,7 @@ pub async fn run_plugin_v3(mut shared: impl XcpPlugin, target_daemon_path: Optio
     loop {
         tracing::debug!("Updating plugin state");
         // Update sources
-        shared.update();
+        shared.update().await;
 
         // Fetch and push new metrics.
         plugin

diff --git a/plugins/xcp-metrics-plugin-common/src/protocol_v2.rs b/plugins/xcp-metrics-plugin-common/src/protocol_v2.rs
@@ -4,12 +4,13 @@ use tokio::{
     fs::{create_dir_all, OpenOptions},
     io::AsyncWriteExt,
 };
+
 use xapi::{
     rpc::{
-        message::parse_http_response,
-        methods::{PluginLocalDeregister, PluginLocalRegister},
+        message::{parse_http_response, RpcKind},
+        methods::rrdd::{PluginLocalDeregister, PluginLocalRegister},
     },
-    METRICS_SHM_PATH,
+    unix::METRICS_SHM_PATH,
 };
 use xcp_metrics_common::rrdd::{
     protocol_common::DataSourceValue,
@@ -67,11 +68,12 @@ impl RrddPlugin {
             uid: self.uid.to_string(),
         };
 
-        let response = xapi::send_xmlrpc_to(
+        let response = xapi::unix::send_rpc_to(
             &self.target_daemon_path,
             "POST",
             &request,
             &self.uid, /* use uid as user-agent */
+            RpcKind::XmlRpc,
         )
         .await
         .map_err(|e| {
@@ -146,16 +148,17 @@ impl RrddPlugin {
             uid: self.uid.to_string(),
         };
 
-        match xapi::send_xmlrpc_to(
+        match xapi::unix::send_rpc_to(
             &self.target_daemon_path,
             "POST",
             &request,
             &self.uid, /* use uid as user-agent */
+            RpcKind::XmlRpc,
         )
         .await
         {
             Ok(response) => {
-                tracing::info!("RPC Response: {:?}", parse_http_response(response).await);
+                tracing::info!("RPC Response: {:?}", parse_http_response(response).await)
             }
             Err(e) => {
                 tracing::error!("Unable to unregister plugin ({e})")
@@ -164,7 +167,7 @@ impl RrddPlugin {
 
         // Delete plugin file.
         if let Err(e) = tokio::fs::remove_file(self.metrics_path).await {
-            tracing::warn!("Unable to remove plugin file: {e}");
+            tracing::warn!("Unable to remove plugin file: {e}")
         }
     }
 }
diff --git a/plugins/xcp-metrics-plugin-common/src/protocol_v3/mod.rs b/plugins/xcp-metrics-plugin-common/src/protocol_v3/mod.rs
@@ -4,12 +4,13 @@ pub mod utils;
 use std::path::{Path, PathBuf};
 
 use tokio::fs::{create_dir_all, OpenOptions};
+
 use xapi::{
     rpc::{
-        message::parse_http_response,
-        methods::{PluginMetricsDeregister, PluginMetricsRegister},
+        message::{parse_http_response, RpcKind},
+        methods::rrdd::{PluginMetricsDeregister, PluginMetricsRegister},
     },
-    METRICS_SHM_PATH,
+    unix::METRICS_SHM_PATH,
 };
 use xcp_metrics_common::{metrics::MetricSet, protocol_v3};
 
@@ -64,11 +65,12 @@ impl MetricsPlugin {
             uid: self.uid.to_string(),
         };
 
-        let response = xapi::send_xmlrpc_to(
+        let response = xapi::unix::send_rpc_to(
             &self.target_daemon_path,
             "POST",
             &request,
             &self.uid, /* use uid as user-agent */
+            RpcKind::JsonRpc,
         )
         .await
         .map_err(|e| {
@@ -92,19 +94,20 @@ impl MetricsPlugin {
             uid: self.uid.to_string(),
         };
 
-        match xapi::send_xmlrpc_to(
+        match xapi::unix::send_rpc_to(
             &self.target_daemon_path,
             "POST",
             &request,
             &self.uid, /* use uid as user-agent */
+            RpcKind::JsonRpc,
         )
         .await
         {
             Ok(response) => {
-                tracing::debug!("RPC Response: {:?}", parse_http_response(response).await);
+                tracing::debug!("RPC Response: {:?}", parse_http_response(response).await)
             }
             Err(e) => {
-                tracing::error!("Unable to unregister plugin ({e})");
+                tracing::error!("Unable to unregister plugin ({e})")
             }
         }
 

diff --git a/plugins/xcp-metrics-plugin-common/src/protocol_v3/utils.rs b/plugins/xcp-metrics-plugin-common/src/protocol_v3/utils.rs
@@ -3,7 +3,7 @@
 use std::{collections::HashMap, time::SystemTime};
 
 use xcp_metrics_common::metrics::{
-    Label, Metric, MetricFamily, MetricPoint, MetricSet, MetricType, MetricValue,
+    Label, Metric, MetricFamily, MetricSet, MetricType, MetricValue, MetricValue,
 };
 
 #[derive(Clone, Debug)]
@@ -41,6 +41,7 @@ impl From<SimpleMetricFamily> for MetricFamily {
         }: SimpleMetricFamily,
     ) -> Self {
         Self {
+            reference_count: 1,
             metric_type,
             unit,
             help,
@@ -62,11 +63,7 @@ impl From<SimpleMetric> for Metric {
     fn from(SimpleMetric { labels, value }: SimpleMetric) -> Self {
         Self {
             labels: labels.into_boxed_slice(),
-            metrics_point: vec![MetricPoint {
-                value,
-                timestamp: SystemTime::now(),
-            }]
-            .into_boxed_slice(),
+            value,
         }
     }
 }
diff --git a/plugins/xcp-metrics-plugin-xenstored/Cargo.toml b/plugins/xcp-metrics-plugin-xenstored/Cargo.toml
@@ -10,18 +10,22 @@ license = "AGPL-3.0-only"
 
 [dependencies]
 xcp-metrics-common = { path = "../../xcp-metrics-common" }
-xcp-metrics-plugin-common = { path = "../xcp-metrics-plugin-common", features = ["xenstore-wip"] }
 
-uuid = "1.4"
+uuid = "1.11"
 anyhow = "1.0"
-dashmap = "5.5"
 futures = "0.3"
+async-stream = "0.3"
 
 tokio = { version = "1", features = ["full"] }
 
 tracing = "0.1"
 tracing-subscriber = "0.3"
 
+xenstore-rs = { version = "0.8", features = ["async-tokio"] }
+smol_str = { version = "0.3" }
+radix_trie = "0.2"
+enum_dispatch = "0.3"
+
 [dependencies.clap]
 version = "4.3"
 features = ["derive"]
diff --git a/plugins/xcp-metrics-plugin-xenstored/src/main.rs b/plugins/xcp-metrics-plugin-xenstored/src/main.rs
@@ -2,13 +2,9 @@ mod plugin;
 
 use clap::{command, Parser};
 use std::path::PathBuf;
-
-use xcp_metrics_plugin_common::{
-    plugin::run_hybrid,
-    xenstore::xs::{Xs, XsOpenFlags},
-};
-
-use plugin::XenStorePlugin;
+use tokio::net::UnixStream;
+use xcp_metrics_common::protocol::METRICS_SOCKET_PATH;
+use xenstore_rs::tokio::XsTokio;
 
 /// xcp-metrics XenStore plugin.
 #[derive(Clone, Parser, Debug)]
@@ -21,10 +17,6 @@ struct Args {
     /// Target daemon.
     #[arg(short, long)]
     target: Option<PathBuf>,
-
-    /// Used protocol
-    #[arg(short, long)]
-    protocol: Option<u32>,
 }
 
 #[tokio::main]
@@ -39,18 +31,23 @@ async fn main() {
 
     tracing::subscriber::set_global_default(text_subscriber).unwrap();
 
-    let xs = match Xs::new(XsOpenFlags::ReadOnly) {
+    let rpc_stream = match UnixStream::connect(METRICS_SOCKET_PATH).await {
+        Ok(stream) => stream,
+        Err(e) => {
+            tracing::error!("Unable to connect to xcp-metrics: {e}");
+            return;
+        }
+    };
+
+    let xs = match XsTokio::new().await {
         Ok(xs) => xs,
         Err(e) => {
-            tracing::error!("Unable to initialize XenStore {e}");
+            tracing::error!("Unable to initialize XenStore: {e}");
             return;
         }
     };
 
-    run_hybrid(
-        XenStorePlugin::new(&xs),
-        args.target.as_deref(),
-        args.protocol,
-    )
-    .await;
+    if let Err(e) = plugin::run_plugin(rpc_stream, xs).await {
+        tracing::error!("Plugin failure {e}");
+    }
 }