* [pbs-devel] [PATCH proxmox/proxmox-backup v3] add metrics server capability
@ 2021-12-17 8:09 Dominik Csapak
2021-12-17 8:09 ` [pbs-devel] [PATCH proxmox v3 1/3] proxmox-sys: make some structs serializable Dominik Csapak
` (8 more replies)
0 siblings, 9 replies; 13+ messages in thread
From: Dominik Csapak @ 2021-12-17 8:09 UTC (permalink / raw)
To: pbs-devel
this series adds support for exporting metrics data to external
metric servers.
for now this includes only data we gather for RRD, though it should
not be hard to extend that functionality
also only influxdb (udp/http(s)) is currently supported, but it should
also not be too hard to include more options here
i did not include gui/cli patches yet, as i find the
proxmox-backup-manager options are already too much and i waited for
the gui for some feedback.
for testing, the metric servers can be added either by
calling 'proxmox-backup debug api ...' or by manually editing the
file
ofc, proxmox-backup depends on bumped versions of the proxmox-* crates
changes from v2:
* rebase on master
* rustfmt
* clippy (fixed not everything)
* renamed DiskUsage in proxmox-sys and added some more fields
* added 'enable' property for the config (like we have in pve)
* subtracted 50bytes from mtu in the udp variant (for ip header)
changes from v1:
* fixed ipv6 support for udp (tested it this time ;) )
* dropped the 'flush' functionality of the MetricsChannel, but kept the
wrapper struct: it did not do what i intended, and after rethinking it,
turns out it's not necessary (as we autoflush when the data gets to large,
or when we close the channel). kept the struct so that the interface
can stay the same even if we want to implement a manual flush in the future
* improved the influxdb line formatter
* removed variables like 'names2' by reorganizing the code
* used Arc::clone(&foo) instead of foo.clone() (better visibilty)
* used CamelCase for the DeletableProperties
proxmox:
Dominik Csapak (3):
proxmox-sys: make some structs serializable
proxmox-sys: add FileSystemInformation struct and helper
proxmox-metrics: implement metrics server client code
Cargo.toml | 1 +
proxmox-metrics/Cargo.toml | 20 +++++
proxmox-metrics/debian/changelog | 5 ++
proxmox-metrics/debian/copyright | 16 ++++
proxmox-metrics/debian/debcargo.toml | 7 ++
proxmox-metrics/src/influxdb/http.rs | 122 ++++++++++++++++++++++++++
proxmox-metrics/src/influxdb/mod.rs | 7 ++
proxmox-metrics/src/influxdb/udp.rs | 94 ++++++++++++++++++++
proxmox-metrics/src/influxdb/utils.rs | 50 +++++++++++
proxmox-metrics/src/lib.rs | 89 +++++++++++++++++++
proxmox-sys/Cargo.toml | 1 +
proxmox-sys/src/fs/mod.rs | 39 ++++++++
proxmox-sys/src/linux/procfs/mod.rs | 7 +-
13 files changed, 455 insertions(+), 3 deletions(-)
create mode 100644 proxmox-metrics/Cargo.toml
create mode 100644 proxmox-metrics/debian/changelog
create mode 100644 proxmox-metrics/debian/copyright
create mode 100644 proxmox-metrics/debian/debcargo.toml
create mode 100644 proxmox-metrics/src/influxdb/http.rs
create mode 100644 proxmox-metrics/src/influxdb/mod.rs
create mode 100644 proxmox-metrics/src/influxdb/udp.rs
create mode 100644 proxmox-metrics/src/influxdb/utils.rs
create mode 100644 proxmox-metrics/src/lib.rs
proxmox-backup:
Dominik Csapak (6):
use 'fs_info' from proxmox-sys
pbs-api-types: add metrics api types
pbs-config: add metrics config class
backup-proxy: decouple stats gathering from rrd update
proxmox-backup-proxy: send metrics to configured metrics server
api: add metricserver endpoints
Cargo.toml | 1 +
pbs-api-types/src/lib.rs | 2 +
pbs-api-types/src/metrics.rs | 147 ++++++++
pbs-config/Cargo.toml | 1 +
pbs-config/src/lib.rs | 1 +
pbs-config/src/metrics.rs | 128 +++++++
src/api2/admin/datastore.rs | 4 +-
src/api2/config/metricserver/influxdbhttp.rs | 275 +++++++++++++++
src/api2/config/metricserver/influxdbudp.rs | 245 +++++++++++++
src/api2/config/metricserver/mod.rs | 16 +
src/api2/config/mod.rs | 2 +
src/api2/node/status.rs | 10 +-
src/api2/status.rs | 4 +-
src/bin/proxmox-backup-proxy.rs | 346 +++++++++++++++----
src/tools/disks/mod.rs | 21 +-
15 files changed, 1105 insertions(+), 98 deletions(-)
create mode 100644 pbs-api-types/src/metrics.rs
create mode 100644 pbs-config/src/metrics.rs
create mode 100644 src/api2/config/metricserver/influxdbhttp.rs
create mode 100644 src/api2/config/metricserver/influxdbudp.rs
create mode 100644 src/api2/config/metricserver/mod.rs
--
2.30.2
^ permalink raw reply [flat|nested] 13+ messages in thread
* [pbs-devel] [PATCH proxmox v3 1/3] proxmox-sys: make some structs serializable
2021-12-17 8:09 [pbs-devel] [PATCH proxmox/proxmox-backup v3] add metrics server capability Dominik Csapak
@ 2021-12-17 8:09 ` Dominik Csapak
2021-12-17 8:09 ` [pbs-devel] [PATCH proxmox v3 2/3] proxmox-sys: add FileSystemInformation struct and helper Dominik Csapak
` (7 subsequent siblings)
8 siblings, 0 replies; 13+ messages in thread
From: Dominik Csapak @ 2021-12-17 8:09 UTC (permalink / raw)
To: pbs-devel
we already depend on serde anyway, and this makes gathering structs a
bit more comfortable
Signed-off-by: Dominik Csapak <d.csapak@proxmox.com>
---
proxmox-sys/Cargo.toml | 1 +
proxmox-sys/src/linux/procfs/mod.rs | 7 ++++---
2 files changed, 5 insertions(+), 3 deletions(-)
diff --git a/proxmox-sys/Cargo.toml b/proxmox-sys/Cargo.toml
index 80d39eb..bfcd388 100644
--- a/proxmox-sys/Cargo.toml
+++ b/proxmox-sys/Cargo.toml
@@ -17,6 +17,7 @@ log = "0.4"
nix = "0.19.1"
regex = "1.2"
serde_json = "1.0"
+serde = { version = "1.0", features = [ "derive" ] }
zstd = { version = "0.6", features = [ "bindgen" ] }
# Macro crates:
diff --git a/proxmox-sys/src/linux/procfs/mod.rs b/proxmox-sys/src/linux/procfs/mod.rs
index 30b9978..3373dec 100644
--- a/proxmox-sys/src/linux/procfs/mod.rs
+++ b/proxmox-sys/src/linux/procfs/mod.rs
@@ -11,6 +11,7 @@ use std::time::Instant;
use anyhow::{bail, format_err, Error};
use lazy_static::lazy_static;
use nix::unistd::Pid;
+use serde::Serialize;
use crate::fs::file_read_firstline;
@@ -184,7 +185,7 @@ pub fn read_proc_uptime_ticks() -> Result<(u64, u64), Error> {
Ok((up as u64, idle as u64))
}
-#[derive(Debug, Default)]
+#[derive(Debug, Default, Serialize)]
/// The CPU fields from `/proc/stat` with their native time value. Multiply
/// with CLOCK_TICKS to get the real value.
pub struct ProcFsStat {
@@ -407,7 +408,7 @@ fn test_read_proc_stat() {
assert_eq!(stat.iowait_percent, 0.0);
}
-#[derive(Debug)]
+#[derive(Debug, Serialize)]
pub struct ProcFsMemInfo {
pub memtotal: u64,
pub memfree: u64,
@@ -539,7 +540,7 @@ pub fn read_memory_usage() -> Result<ProcFsMemUsage, Error> {
}
}
-#[derive(Debug)]
+#[derive(Debug, Serialize)]
pub struct ProcFsNetDev {
pub device: String,
pub receive: u64,
--
2.30.2
^ permalink raw reply [flat|nested] 13+ messages in thread
* [pbs-devel] [PATCH proxmox v3 2/3] proxmox-sys: add FileSystemInformation struct and helper
2021-12-17 8:09 [pbs-devel] [PATCH proxmox/proxmox-backup v3] add metrics server capability Dominik Csapak
2021-12-17 8:09 ` [pbs-devel] [PATCH proxmox v3 1/3] proxmox-sys: make some structs serializable Dominik Csapak
@ 2021-12-17 8:09 ` Dominik Csapak
2022-01-12 14:10 ` Wolfgang Bumiller
2021-12-17 8:09 ` [pbs-devel] [PATCH proxmox v3 3/3] proxmox-metrics: implement metrics server client code Dominik Csapak
` (6 subsequent siblings)
8 siblings, 1 reply; 13+ messages in thread
From: Dominik Csapak @ 2021-12-17 8:09 UTC (permalink / raw)
To: pbs-devel
code mostly copied from proxmox-backups 'disk_usage'
Signed-off-by: Dominik Csapak <d.csapak@proxmox.com>
---
proxmox-sys/src/fs/mod.rs | 39 +++++++++++++++++++++++++++++++++++++++
1 file changed, 39 insertions(+)
diff --git a/proxmox-sys/src/fs/mod.rs b/proxmox-sys/src/fs/mod.rs
index 9fe333b..f8a7744 100644
--- a/proxmox-sys/src/fs/mod.rs
+++ b/proxmox-sys/src/fs/mod.rs
@@ -102,3 +102,42 @@ impl CreateOptions {
*/
}
+/// Information about a mounted file system from statfs64 syscall
+pub struct FileSystemInformation {
+ /// total bytes available
+ pub total: u64,
+ /// bytes used
+ pub used: u64,
+ /// bytes available to an unprivileged user
+ pub available: u64,
+ /// total number of inodes
+ pub total_inodes: u64,
+ /// free number of inodes
+ pub free_inodes: u64,
+ /// the type of the filesystem (see statfs64(2))
+ pub fs_type: i64,
+ /// the filesystem id
+ pub fs_id: libc::fsid_t,
+}
+
+/// Get file system information from path
+pub fn fs_info(path: &std::path::Path) -> Result<FileSystemInformation, Error> {
+ let mut stat: libc::statfs64 = unsafe { std::mem::zeroed() };
+
+ use nix::NixPath;
+
+ let res = path.with_nix_path(|cstr| unsafe { libc::statfs64(cstr.as_ptr(), &mut stat) })?;
+ nix::errno::Errno::result(res)?;
+
+ let bsize = stat.f_bsize as u64;
+
+ Ok(FileSystemInformation {
+ total: stat.f_blocks * bsize,
+ used: (stat.f_blocks - stat.f_bfree) * bsize,
+ available: stat.f_bavail * bsize,
+ total_inodes: stat.f_files,
+ free_inodes: stat.f_ffree,
+ fs_type: stat.f_type,
+ fs_id: stat.f_fsid,
+ })
+}
--
2.30.2
^ permalink raw reply [flat|nested] 13+ messages in thread
* [pbs-devel] [PATCH proxmox v3 3/3] proxmox-metrics: implement metrics server client code
2021-12-17 8:09 [pbs-devel] [PATCH proxmox/proxmox-backup v3] add metrics server capability Dominik Csapak
2021-12-17 8:09 ` [pbs-devel] [PATCH proxmox v3 1/3] proxmox-sys: make some structs serializable Dominik Csapak
2021-12-17 8:09 ` [pbs-devel] [PATCH proxmox v3 2/3] proxmox-sys: add FileSystemInformation struct and helper Dominik Csapak
@ 2021-12-17 8:09 ` Dominik Csapak
2022-01-12 14:36 ` Wolfgang Bumiller
2021-12-17 8:09 ` [pbs-devel] [PATCH proxmox-backup v3 1/6] use 'fs_info' from proxmox-sys Dominik Csapak
` (5 subsequent siblings)
8 siblings, 1 reply; 13+ messages in thread
From: Dominik Csapak @ 2021-12-17 8:09 UTC (permalink / raw)
To: pbs-devel
influxdb (udp + http(s)) only for now
general architecture looks as follows:
"new" returns a MetricsChannel and a Future
the channels can be used to push data in (it flushes automatically if
it would be over the configured size (mtu/max_body_size))
and the future must be polled to actually send data to the servers.
so most often it would look like this:
let (future, channel) = InfluxDbHttp::new(..params..)?;
let handle = tokio::spawn(future);
channel.send_data(...).await?;
handle.await?;
when all channels go out of scope, all remaining data in the channel
will be read and sent to the server
Signed-off-by: Dominik Csapak <d.csapak@proxmox.com>
---
Cargo.toml | 1 +
proxmox-metrics/Cargo.toml | 20 +++++
proxmox-metrics/debian/changelog | 5 ++
proxmox-metrics/debian/copyright | 16 ++++
proxmox-metrics/debian/debcargo.toml | 7 ++
proxmox-metrics/src/influxdb/http.rs | 122 ++++++++++++++++++++++++++
proxmox-metrics/src/influxdb/mod.rs | 7 ++
proxmox-metrics/src/influxdb/udp.rs | 94 ++++++++++++++++++++
proxmox-metrics/src/influxdb/utils.rs | 50 +++++++++++
proxmox-metrics/src/lib.rs | 89 +++++++++++++++++++
10 files changed, 411 insertions(+)
create mode 100644 proxmox-metrics/Cargo.toml
create mode 100644 proxmox-metrics/debian/changelog
create mode 100644 proxmox-metrics/debian/copyright
create mode 100644 proxmox-metrics/debian/debcargo.toml
create mode 100644 proxmox-metrics/src/influxdb/http.rs
create mode 100644 proxmox-metrics/src/influxdb/mod.rs
create mode 100644 proxmox-metrics/src/influxdb/udp.rs
create mode 100644 proxmox-metrics/src/influxdb/utils.rs
create mode 100644 proxmox-metrics/src/lib.rs
diff --git a/Cargo.toml b/Cargo.toml
index 8f85e08..4a458d2 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -6,6 +6,7 @@ members = [
"proxmox-http",
"proxmox-io",
"proxmox-lang",
+ "proxmox-metrics",
"proxmox-router",
"proxmox-schema",
"proxmox-serde",
diff --git a/proxmox-metrics/Cargo.toml b/proxmox-metrics/Cargo.toml
new file mode 100644
index 0000000..9ac50fe
--- /dev/null
+++ b/proxmox-metrics/Cargo.toml
@@ -0,0 +1,20 @@
+[package]
+name = "proxmox-metrics"
+version = "0.1.0"
+authors = ["Proxmox Support Team <support@proxmox.com>"]
+edition = "2018"
+license = "AGPL-3"
+description = "Metrics Server export utilitites"
+
+exclude = [ "debian" ]
+
+[dependencies]
+anyhow = "1.0"
+tokio = { version = "1.0", features = [ "net", "sync" ] }
+futures = "0.3"
+serde = "1.0"
+serde_json = "1.0"
+http = "0.2"
+hyper = "0.14"
+openssl = "0.10"
+proxmox-http = { path = "../proxmox-http", features = [ "client" ], version = "0.6" }
diff --git a/proxmox-metrics/debian/changelog b/proxmox-metrics/debian/changelog
new file mode 100644
index 0000000..c02803b
--- /dev/null
+++ b/proxmox-metrics/debian/changelog
@@ -0,0 +1,5 @@
+rust-proxmox-metrics (0.1.0-1) unstable; urgency=medium
+
+ * initial package
+
+ -- Proxmox Support Team <support@proxmox.com> Tue, 14 Dec 2021 08:56:54 +0100
diff --git a/proxmox-metrics/debian/copyright b/proxmox-metrics/debian/copyright
new file mode 100644
index 0000000..5661ef6
--- /dev/null
+++ b/proxmox-metrics/debian/copyright
@@ -0,0 +1,16 @@
+Copyright (C) 2021 Proxmox Server Solutions GmbH
+
+This software is written by Proxmox Server Solutions GmbH <support@proxmox.com>
+
+This program is free software: you can redistribute it and/or modify
+it under the terms of the GNU Affero General Public License as published by
+the Free Software Foundation, either version 3 of the License, or
+(at your option) any later version.
+
+This program 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 Affero General Public License for more details.
+
+You should have received a copy of the GNU Affero General Public License
+along with this program. If not, see <http://www.gnu.org/licenses/>.
diff --git a/proxmox-metrics/debian/debcargo.toml b/proxmox-metrics/debian/debcargo.toml
new file mode 100644
index 0000000..b7864cd
--- /dev/null
+++ b/proxmox-metrics/debian/debcargo.toml
@@ -0,0 +1,7 @@
+overlay = "."
+crate_src_path = ".."
+maintainer = "Proxmox Support Team <support@proxmox.com>"
+
+[source]
+vcs_git = "git://git.proxmox.com/git/proxmox.git"
+vcs_browser = "https://git.proxmox.com/?p=proxmox.git"
diff --git a/proxmox-metrics/src/influxdb/http.rs b/proxmox-metrics/src/influxdb/http.rs
new file mode 100644
index 0000000..df6c6a5
--- /dev/null
+++ b/proxmox-metrics/src/influxdb/http.rs
@@ -0,0 +1,122 @@
+use std::sync::Arc;
+
+use anyhow::{bail, Error};
+use hyper::Body;
+use openssl::ssl::{SslConnector, SslMethod, SslVerifyMode};
+use tokio::sync::mpsc;
+
+use proxmox_http::client::{SimpleHttp, SimpleHttpOptions};
+
+use crate::influxdb::utils;
+use crate::{MetricsChannel, MetricsData, MetricsServerFuture};
+
+pub struct InfluxDbHttp {
+ client: SimpleHttp,
+ _healthuri: http::Uri,
+ writeuri: http::Uri,
+ token: Option<String>,
+ max_body_size: usize,
+ data: String,
+ channel: mpsc::Receiver<Arc<MetricsData>>,
+}
+
+impl InfluxDbHttp {
+ pub fn new(
+ https: bool,
+ host: &str,
+ port: u16,
+ organization: &str,
+ bucket: &str,
+ token: Option<&str>,
+ verify_tls: bool,
+ max_body_size: usize,
+ ) -> Result<(MetricsServerFuture, MetricsChannel), Error> {
+ let (tx, rx) = mpsc::channel(1024);
+
+ let client = if verify_tls {
+ SimpleHttp::with_options(SimpleHttpOptions::default())
+ } else {
+ let mut ssl_connector = SslConnector::builder(SslMethod::tls()).unwrap();
+ ssl_connector.set_verify(SslVerifyMode::NONE);
+ SimpleHttp::with_ssl_connector(ssl_connector.build(), SimpleHttpOptions::default())
+ };
+
+ let authority = proxmox_http::uri::build_authority(host, port)?;
+
+ let writeuri = http::uri::Builder::new()
+ .scheme(if https { "https" } else { "http" })
+ .authority(authority.clone())
+ .path_and_query(format!(
+ "/api/v2/write?org={}&bucket={}",
+ organization, bucket
+ ))
+ .build()?;
+
+ let healthuri = http::uri::Builder::new()
+ .scheme(if https { "https" } else { "http" })
+ .authority(authority)
+ .path_and_query("/health")
+ .build()?;
+
+ let this = Self {
+ client,
+ writeuri,
+ _healthuri: healthuri,
+ token: token.map(String::from),
+ max_body_size,
+ data: String::new(),
+ channel: rx,
+ };
+
+ let future = Box::pin(this.finish());
+ let channel = MetricsChannel { channel: tx };
+ Ok((future, channel))
+ }
+
+ async fn add_data(&mut self, data: Arc<MetricsData>) -> Result<(), Error> {
+ let new_data = utils::format_influxdb_line(&data)?;
+
+ if self.data.len() + new_data.len() >= self.max_body_size {
+ self.flush().await?;
+ }
+
+ self.data.push_str(&new_data);
+
+ if self.data.len() >= self.max_body_size {
+ self.flush().await?;
+ }
+
+ Ok(())
+ }
+
+ pub async fn flush(&mut self) -> Result<(), Error> {
+ if self.data.is_empty() {
+ return Ok(());
+ }
+ let mut request = http::Request::builder().method("POST").uri(&self.writeuri);
+
+ if let Some(token) = &self.token {
+ request = request.header("Authorization", format!("Token {}", token));
+ }
+
+ let request = request.body(Body::from(self.data.split_off(0)))?;
+
+ let res = self.client.request(request).await?;
+
+ let status = res.status();
+ if !status.is_success() {
+ bail!("got bad status: {}", status);
+ }
+ Ok(())
+ }
+
+ async fn finish(mut self) -> Result<(), Error> {
+ while let Some(data) = self.channel.recv().await {
+ self.add_data(data).await?;
+ }
+
+ self.flush().await?;
+
+ Ok(())
+ }
+}
diff --git a/proxmox-metrics/src/influxdb/mod.rs b/proxmox-metrics/src/influxdb/mod.rs
new file mode 100644
index 0000000..26a715c
--- /dev/null
+++ b/proxmox-metrics/src/influxdb/mod.rs
@@ -0,0 +1,7 @@
+mod http;
+pub use self::http::*;
+
+mod udp;
+pub use udp::*;
+
+pub mod utils;
diff --git a/proxmox-metrics/src/influxdb/udp.rs b/proxmox-metrics/src/influxdb/udp.rs
new file mode 100644
index 0000000..c4187a6
--- /dev/null
+++ b/proxmox-metrics/src/influxdb/udp.rs
@@ -0,0 +1,94 @@
+use std::sync::Arc;
+
+use anyhow::Error;
+use tokio::net::UdpSocket;
+use tokio::sync::mpsc;
+
+use crate::influxdb::utils;
+use crate::{MetricsChannel, MetricsData, MetricsServerFuture};
+
+pub struct InfluxDbUdp {
+ address: String,
+ conn: Option<tokio::net::UdpSocket>,
+ ipv6: bool,
+ mtu: u16,
+ data: String,
+ channel: mpsc::Receiver<Arc<MetricsData>>,
+}
+
+impl InfluxDbUdp {
+ pub fn new(host: &str, port: u16, mtu: Option<u16>) -> (MetricsServerFuture, MetricsChannel) {
+ let (tx, rx) = mpsc::channel(1024);
+ let ipv6 = host.contains(':');
+
+ let address = if ipv6 && host.len() > 3 && &host[0..1] != "[" {
+ format!("[{}]:{}", host, port)
+ } else {
+ format!("{}:{}", host, port)
+ };
+
+ let this = Self {
+ address,
+ ipv6,
+ conn: None,
+ // empty ipv6 udp package needs 48 bytes, subtract 50 for safety
+ mtu: mtu.unwrap_or(1500) - 50,
+ data: String::new(),
+ channel: rx,
+ };
+
+ let future = Box::pin(this.finish());
+
+ let channel = MetricsChannel { channel: tx };
+
+ (future, channel)
+ }
+
+ async fn connect(&mut self) -> Result<tokio::net::UdpSocket, Error> {
+ let conn = if self.ipv6 {
+ UdpSocket::bind("[::]:0").await?
+ } else {
+ UdpSocket::bind("0.0.0.0:0").await?
+ };
+ let addr = self.address.clone();
+ conn.connect(addr).await?;
+ Ok(conn)
+ }
+
+ async fn add_data(&mut self, data: Arc<MetricsData>) -> Result<(), Error> {
+ let new_data = utils::format_influxdb_line(&data)?;
+
+ if self.data.len() + new_data.len() >= (self.mtu as usize) {
+ self.flush().await?;
+ }
+
+ self.data.push_str(&new_data);
+
+ if self.data.len() >= (self.mtu as usize) {
+ self.flush().await?;
+ }
+
+ Ok(())
+ }
+
+ async fn flush(&mut self) -> Result<(), Error> {
+ let conn = match self.conn.take() {
+ Some(conn) => conn,
+ None => self.connect().await?,
+ };
+
+ conn.send(self.data.split_off(0).as_bytes()).await?;
+ self.conn = Some(conn);
+ Ok(())
+ }
+
+ async fn finish(mut self) -> Result<(), Error> {
+ while let Some(data) = self.channel.recv().await {
+ self.add_data(data).await?;
+ }
+
+ self.flush().await?;
+
+ Ok(())
+ }
+}
diff --git a/proxmox-metrics/src/influxdb/utils.rs b/proxmox-metrics/src/influxdb/utils.rs
new file mode 100644
index 0000000..3507e61
--- /dev/null
+++ b/proxmox-metrics/src/influxdb/utils.rs
@@ -0,0 +1,50 @@
+use std::fmt::Write;
+
+use anyhow::{bail, Error};
+use serde_json::Value;
+
+use crate::MetricsData;
+
+pub(crate) fn format_influxdb_line(data: &MetricsData) -> Result<String, Error> {
+ if !data.values.is_object() {
+ bail!("invalid data");
+ }
+
+ let mut line = escape_measurement(&data.measurement);
+
+ for (key, value) in &data.tags {
+ write!(line, ",{}={}", escape_key(key), escape_key(value))?;
+ }
+
+ line.push(' ');
+
+ let mut first = true;
+ for (key, value) in data.values.as_object().unwrap().iter() {
+ match value {
+ Value::Object(_) => bail!("objects not supported"),
+ Value::Array(_) => bail!("arrays not supported"),
+ _ => {}
+ }
+ if !first {
+ line.push(',');
+ }
+ first = false;
+ write!(line, "{}={}", escape_key(key), value.to_string())?;
+ }
+
+ // nanosecond precision
+ writeln!(line, " {}", data.ctime * 1_000_000_000)?;
+
+ Ok(line)
+}
+
+fn escape_key(key: &str) -> String {
+ let key = key.replace(',', "\\,");
+ let key = key.replace('=', "\\=");
+ key.replace(' ', "\\ ")
+}
+
+fn escape_measurement(measurement: &str) -> String {
+ let measurement = measurement.replace(',', "\\,");
+ measurement.replace(' ', "\\ ")
+}
diff --git a/proxmox-metrics/src/lib.rs b/proxmox-metrics/src/lib.rs
new file mode 100644
index 0000000..ba018fb
--- /dev/null
+++ b/proxmox-metrics/src/lib.rs
@@ -0,0 +1,89 @@
+use std::collections::HashMap;
+use std::pin::Pin;
+use std::sync::Arc;
+
+use anyhow::{format_err, Error};
+use serde::Serialize;
+use serde_json::Value;
+use tokio::sync::mpsc;
+
+pub mod influxdb;
+
+#[derive(Clone)]
+/// Structured data for the metric server
+pub struct MetricsData {
+ /// The category of measurements
+ pub measurement: String,
+ /// A list of to attach to the measurements
+ pub tags: HashMap<String, String>,
+ /// The actual values to send. Only plain (not-nested) objects are supported at the moment.
+ pub values: Value,
+ /// The time of the measurement
+ pub ctime: i64,
+}
+
+impl MetricsData {
+ /// Convenient helper to create from references
+ pub fn new<V: Serialize>(
+ measurement: &str,
+ tags: &[(&str, &str)],
+ ctime: i64,
+ values: V,
+ ) -> Result<Self, Error> {
+ let mut new_tags = HashMap::new();
+ for (key, value) in tags {
+ new_tags.insert(key.to_string(), value.to_string());
+ }
+
+ Ok(Self {
+ measurement: measurement.to_string(),
+ tags: new_tags,
+ values: serde_json::to_value(values)?,
+ ctime,
+ })
+ }
+}
+
+pub type MetricsServerFuture =
+ Pin<Box<dyn std::future::Future<Output = Result<(), Error>> + Send + 'static>>;
+
+#[derive(Clone)]
+/// A channel to send data to the metric server
+pub struct MetricsChannel {
+ pub(crate) channel: mpsc::Sender<Arc<MetricsData>>,
+}
+
+impl MetricsChannel {
+ /// Queues the given data to the metric server
+ pub async fn send_data(&self, data: Arc<MetricsData>) -> Result<(), Error> {
+ // return ok if we got no data to send
+ if let Value::Object(map) = &data.values {
+ if map.is_empty() {
+ return Ok(());
+ }
+ }
+ self.channel
+ .send(data)
+ .await
+ .map_err(|_| format_err!("receiver side closed"))?;
+ Ok(())
+ }
+}
+
+/// Helper to send a list of MetricsData to a list of MetricsChannels
+pub async fn send_data_to_channels(
+ values: &[Arc<MetricsData>],
+ channels: &[MetricsChannel],
+) -> Vec<Result<(), Error>> {
+ let mut futures = Vec::with_capacity(channels.len());
+ for channel in channels {
+ futures.push(async move {
+ for data in values {
+ channel.send_data(Arc::clone(data)).await?
+ }
+ Ok::<(), Error>(())
+ });
+ }
+
+ futures::future::join_all(futures).await
+}
--
2.30.2
^ permalink raw reply [flat|nested] 13+ messages in thread
* [pbs-devel] [PATCH proxmox-backup v3 1/6] use 'fs_info' from proxmox-sys
2021-12-17 8:09 [pbs-devel] [PATCH proxmox/proxmox-backup v3] add metrics server capability Dominik Csapak
` (2 preceding siblings ...)
2021-12-17 8:09 ` [pbs-devel] [PATCH proxmox v3 3/3] proxmox-metrics: implement metrics server client code Dominik Csapak
@ 2021-12-17 8:09 ` Dominik Csapak
2021-12-17 8:09 ` [pbs-devel] [PATCH proxmox-backup v3 2/6] pbs-api-types: add metrics api types Dominik Csapak
` (4 subsequent siblings)
8 siblings, 0 replies; 13+ messages in thread
From: Dominik Csapak @ 2021-12-17 8:09 UTC (permalink / raw)
To: pbs-devel
as replacement for 'disk_usage' in tools. also remove that there since
we do not need it anymore
Signed-off-by: Dominik Csapak <d.csapak@proxmox.com>
---
src/api2/admin/datastore.rs | 4 ++--
src/api2/node/status.rs | 10 ++++++++--
src/api2/status.rs | 4 ++--
src/bin/proxmox-backup-proxy.rs | 4 ++--
src/tools/disks/mod.rs | 21 +--------------------
5 files changed, 15 insertions(+), 28 deletions(-)
diff --git a/src/api2/admin/datastore.rs b/src/api2/admin/datastore.rs
index b653f906..029153b3 100644
--- a/src/api2/admin/datastore.rs
+++ b/src/api2/admin/datastore.rs
@@ -612,7 +612,7 @@ pub fn status(
rpcenv: &mut dyn RpcEnvironment,
) -> Result<DataStoreStatus, Error> {
let datastore = DataStore::lookup_datastore(&store)?;
- let storage = crate::tools::disks::disk_usage(&datastore.base_path())?;
+ let storage = proxmox_sys::fs::fs_info(&datastore.base_path())?;
let (counts, gc_status) = if verbose {
let auth_id: Authid = rpcenv.get_auth_id().unwrap().parse()?;
let user_info = CachedUserInfo::new()?;
@@ -635,7 +635,7 @@ pub fn status(
Ok(DataStoreStatus {
total: storage.total,
used: storage.used,
- avail: storage.avail,
+ avail: storage.available,
gc_status,
counts,
})
diff --git a/src/api2/node/status.rs b/src/api2/node/status.rs
index 9559dda6..fb394fbf 100644
--- a/src/api2/node/status.rs
+++ b/src/api2/node/status.rs
@@ -9,7 +9,7 @@ use proxmox_sys::linux::procfs;
use proxmox_router::{ApiMethod, Router, RpcEnvironment, Permission};
use proxmox_schema::api;
-use pbs_api_types::{NODE_SCHEMA, NodePowerCommand, PRIV_SYS_AUDIT, PRIV_SYS_POWER_MANAGEMENT};
+use pbs_api_types::{NODE_SCHEMA, NodePowerCommand, PRIV_SYS_AUDIT, PRIV_SYS_POWER_MANAGEMENT, StorageStatus};
use crate::api2::types::{
NodeCpuInformation, NodeStatus, NodeMemoryCounters, NodeSwapCounters, NodeInformation,
@@ -77,10 +77,16 @@ fn get_status(
uname.version()
);
+ let disk = proxmox_sys::fs::fs_info(Path::new("/"))?;
+
Ok(NodeStatus {
memory,
swap,
- root: crate::tools::disks::disk_usage(Path::new("/"))?,
+ root: StorageStatus {
+ total: disk.total,
+ used: disk.used,
+ avail: disk.available,
+ },
uptime: procfs::read_proc_uptime()?.0 as u64,
loadavg,
kversion,
diff --git a/src/api2/status.rs b/src/api2/status.rs
index 7f50914b..9424340c 100644
--- a/src/api2/status.rs
+++ b/src/api2/status.rs
@@ -110,13 +110,13 @@ pub fn datastore_status(
continue;
}
};
- let status = crate::tools::disks::disk_usage(&datastore.base_path())?;
+ let status = proxmox_sys::fs::fs_info(&datastore.base_path())?;
let mut entry = json!({
"store": store,
"total": status.total,
"used": status.used,
- "avail": status.avail,
+ "avail": status.available,
"gc-status": datastore.last_gc_status(),
});
diff --git a/src/bin/proxmox-backup-proxy.rs b/src/bin/proxmox-backup-proxy.rs
index 07a53687..fbc5f65c 100644
--- a/src/bin/proxmox-backup-proxy.rs
+++ b/src/bin/proxmox-backup-proxy.rs
@@ -1055,7 +1055,7 @@ fn check_schedule(worker_type: &str, event_str: &str, id: &str) -> bool {
fn gather_disk_stats(disk_manager: Arc<DiskManage>, path: &Path, rrd_prefix: &str) {
- match proxmox_backup::tools::disks::disk_usage(path) {
+ match proxmox_sys::fs::fs_info(path) {
Ok(status) => {
let rrd_key = format!("{}/total", rrd_prefix);
rrd_update_gauge(&rrd_key, status.total as f64);
@@ -1063,7 +1063,7 @@ fn gather_disk_stats(disk_manager: Arc<DiskManage>, path: &Path, rrd_prefix: &st
rrd_update_gauge(&rrd_key, status.used as f64);
}
Err(err) => {
- eprintln!("read disk_usage on {:?} failed - {}", path, err);
+ eprintln!("read fs info on {:?} failed - {}", path, err);
}
}
diff --git a/src/tools/disks/mod.rs b/src/tools/disks/mod.rs
index 867aa624..30d9568f 100644
--- a/src/tools/disks/mod.rs
+++ b/src/tools/disks/mod.rs
@@ -19,7 +19,7 @@ use proxmox_sys::linux::procfs::{MountInfo, mountinfo::Device};
use proxmox_sys::{io_bail, io_format_err};
use proxmox_schema::api;
-use pbs_api_types::{BLOCKDEVICE_NAME_REGEX, StorageStatus};
+use pbs_api_types::BLOCKDEVICE_NAME_REGEX;
mod zfs;
pub use zfs::*;
@@ -521,25 +521,6 @@ impl Disk {
}
}
-/// Returns disk usage information (total, used, avail)
-pub fn disk_usage(path: &std::path::Path) -> Result<StorageStatus, Error> {
-
- let mut stat: libc::statfs64 = unsafe { std::mem::zeroed() };
-
- use nix::NixPath;
-
- let res = path.with_nix_path(|cstr| unsafe { libc::statfs64(cstr.as_ptr(), &mut stat) })?;
- nix::errno::Errno::result(res)?;
-
- let bsize = stat.f_bsize as u64;
-
- Ok(StorageStatus{
- total: stat.f_blocks*bsize,
- used: (stat.f_blocks-stat.f_bfree)*bsize,
- avail: stat.f_bavail*bsize,
- })
-}
-
#[api()]
#[derive(Debug, Serialize, Deserialize)]
#[serde(rename_all="lowercase")]
--
2.30.2
^ permalink raw reply [flat|nested] 13+ messages in thread
* [pbs-devel] [PATCH proxmox-backup v3 2/6] pbs-api-types: add metrics api types
2021-12-17 8:09 [pbs-devel] [PATCH proxmox/proxmox-backup v3] add metrics server capability Dominik Csapak
` (3 preceding siblings ...)
2021-12-17 8:09 ` [pbs-devel] [PATCH proxmox-backup v3 1/6] use 'fs_info' from proxmox-sys Dominik Csapak
@ 2021-12-17 8:09 ` Dominik Csapak
2022-01-12 14:40 ` Wolfgang Bumiller
2021-12-17 8:09 ` [pbs-devel] [PATCH proxmox-backup v3 3/6] pbs-config: add metrics config class Dominik Csapak
` (3 subsequent siblings)
8 siblings, 1 reply; 13+ messages in thread
From: Dominik Csapak @ 2021-12-17 8:09 UTC (permalink / raw)
To: pbs-devel
InfluxDbUdp and InfluxDbHttp for now
Signed-off-by: Dominik Csapak <d.csapak@proxmox.com>
---
pbs-api-types/src/lib.rs | 2 +
pbs-api-types/src/metrics.rs | 147 +++++++++++++++++++++++++++++++++++
2 files changed, 149 insertions(+)
create mode 100644 pbs-api-types/src/metrics.rs
diff --git a/pbs-api-types/src/lib.rs b/pbs-api-types/src/lib.rs
index 0a0dd33d..09bd59c8 100644
--- a/pbs-api-types/src/lib.rs
+++ b/pbs-api-types/src/lib.rs
@@ -88,6 +88,8 @@ pub use traffic_control::*;
mod zfs;
pub use zfs::*;
+mod metrics;
+pub use metrics::*;
#[rustfmt::skip]
#[macro_use]
diff --git a/pbs-api-types/src/metrics.rs b/pbs-api-types/src/metrics.rs
new file mode 100644
index 00000000..786abe82
--- /dev/null
+++ b/pbs-api-types/src/metrics.rs
@@ -0,0 +1,147 @@
+use serde::{Deserialize, Serialize};
+
+use crate::{DNS_NAME_OR_IP_SCHEMA, PROXMOX_SAFE_ID_FORMAT, SINGLE_LINE_COMMENT_SCHEMA};
+use proxmox_schema::{api, Schema, StringSchema, Updater};
+
+pub const METRIC_SERVER_ID_SCHEMA: Schema = StringSchema::new("Metrics Server ID.")
+ .format(&PROXMOX_SAFE_ID_FORMAT)
+ .min_length(3)
+ .max_length(32)
+ .schema();
+
+pub const INFLUXDB_BUCKET_SCHEMA: Schema = StringSchema::new("InfluxDB Bucket.")
+ .format(&PROXMOX_SAFE_ID_FORMAT)
+ .min_length(3)
+ .max_length(32)
+ .default("proxmox")
+ .schema();
+
+pub const INFLUXDB_ORGANIZATION_SCHEMA: Schema = StringSchema::new("InfluxDB Organization.")
+ .format(&PROXMOX_SAFE_ID_FORMAT)
+ .min_length(3)
+ .max_length(32)
+ .default("proxmox")
+ .schema();
+
+#[api(
+ properties: {
+ name: {
+ schema: METRIC_SERVER_ID_SCHEMA,
+ },
+ enable: {
+ description: "Enables or disables the metrics server",
+ type: bool,
+ optional: true,
+ default: true,
+ },
+ host: {
+ schema: DNS_NAME_OR_IP_SCHEMA,
+ },
+ mtu: {
+ description: "The MTU",
+ type: u16,
+ optional: true,
+ default: 1500,
+ },
+ comment: {
+ optional: true,
+ schema: SINGLE_LINE_COMMENT_SCHEMA,
+ },
+ },
+)]
+#[derive(Serialize, Deserialize, Updater)]
+#[serde(rename_all = "kebab-case")]
+/// InfluxDB Server (UDP)
+pub struct InfluxDbUdp {
+ #[updater(skip)]
+ pub name: String,
+ #[serde(skip_serializing_if = "Option::is_none")]
+ pub enable: Option<bool>,
+ pub host: String,
+ /// The port
+ pub port: u16,
+ #[serde(skip_serializing_if = "Option::is_none")]
+ pub mtu: Option<u16>,
+ #[serde(skip_serializing_if = "Option::is_none")]
+ pub comment: Option<String>,
+}
+
+#[api(
+ properties: {
+ name: {
+ schema: METRIC_SERVER_ID_SCHEMA,
+ },
+ enable: {
+ description: "Enables or disables the metrics server",
+ type: bool,
+ optional: true,
+ default: true,
+ },
+ host: {
+ schema: DNS_NAME_OR_IP_SCHEMA,
+ },
+ https: {
+ description: "If true, HTTPS is used.",
+ type: bool,
+ optional: true,
+ default: true,
+ },
+ token: {
+ description: "The (optional) API token",
+ type: String,
+ optional: true,
+ },
+ bucket: {
+ schema: INFLUXDB_BUCKET_SCHEMA,
+ optional: true,
+ },
+ organization: {
+ schema: INFLUXDB_ORGANIZATION_SCHEMA,
+ optional: true,
+ },
+ "max-body-size": {
+ description: "The (optional) maximum body size",
+ type: usize,
+ optional: true,
+ default: 25_000_000,
+ },
+ "verify-tls": {
+ description: "If true, the certificate will be validated.",
+ type: bool,
+ optional: true,
+ default: true,
+ },
+ comment: {
+ optional: true,
+ schema: SINGLE_LINE_COMMENT_SCHEMA,
+ },
+ },
+)]
+#[derive(Serialize, Deserialize, Updater)]
+#[serde(rename_all = "kebab-case")]
+/// InfluxDB Server (HTTP(s))
+pub struct InfluxDbHttp {
+ #[updater(skip)]
+ pub name: String,
+ #[serde(skip_serializing_if = "Option::is_none")]
+ pub enable: Option<bool>,
+ pub host: String,
+ #[serde(skip_serializing_if = "Option::is_none")]
+ /// The (optional) port. (defaults: 80 for HTTP, 443 for HTTPS)
+ pub port: Option<u16>,
+ #[serde(skip_serializing_if = "Option::is_none")]
+ pub https: Option<bool>,
+ /// The Optional Token
+ #[serde(skip_serializing_if = "Option::is_none")]
+ pub token: Option<String>,
+ #[serde(skip_serializing_if = "Option::is_none")]
+ pub bucket: Option<String>,
+ #[serde(skip_serializing_if = "Option::is_none")]
+ pub organization: Option<String>,
+ #[serde(skip_serializing_if = "Option::is_none")]
+ pub max_body_size: Option<usize>,
+ #[serde(skip_serializing_if = "Option::is_none")]
+ pub verify_tls: Option<bool>,
+ #[serde(skip_serializing_if = "Option::is_none")]
+ pub comment: Option<String>,
+}
--
2.30.2
^ permalink raw reply [flat|nested] 13+ messages in thread
* [pbs-devel] [PATCH proxmox-backup v3 3/6] pbs-config: add metrics config class
2021-12-17 8:09 [pbs-devel] [PATCH proxmox/proxmox-backup v3] add metrics server capability Dominik Csapak
` (4 preceding siblings ...)
2021-12-17 8:09 ` [pbs-devel] [PATCH proxmox-backup v3 2/6] pbs-api-types: add metrics api types Dominik Csapak
@ 2021-12-17 8:09 ` Dominik Csapak
2021-12-17 8:09 ` [pbs-devel] [PATCH proxmox-backup v3 4/6] backup-proxy: decouple stats gathering from rrd update Dominik Csapak
` (2 subsequent siblings)
8 siblings, 0 replies; 13+ messages in thread
From: Dominik Csapak @ 2021-12-17 8:09 UTC (permalink / raw)
To: pbs-devel
a section config like in pve
Signed-off-by: Dominik Csapak <d.csapak@proxmox.com>
---
Cargo.toml | 1 +
pbs-config/Cargo.toml | 1 +
pbs-config/src/lib.rs | 1 +
pbs-config/src/metrics.rs | 128 ++++++++++++++++++++++++++++++++++++++
4 files changed, 131 insertions(+)
create mode 100644 pbs-config/src/metrics.rs
diff --git a/Cargo.toml b/Cargo.toml
index 0debfa93..9721fc6e 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -96,6 +96,7 @@ pxar = { version = "0.10.1", features = [ "tokio-io" ] }
proxmox-http = { version = "0.6", features = [ "client", "http-helpers", "websocket" ] }
proxmox-io = "1"
proxmox-lang = "1"
+proxmox-metrics = "0.1"
proxmox-router = { version = "1.1", features = [ "cli" ] }
proxmox-schema = { version = "1.1", features = [ "api-macro" ] }
proxmox-section-config = "1"
diff --git a/pbs-config/Cargo.toml b/pbs-config/Cargo.toml
index 7c3b31cb..35b70c1d 100644
--- a/pbs-config/Cargo.toml
+++ b/pbs-config/Cargo.toml
@@ -25,6 +25,7 @@ proxmox-time = "1"
proxmox-serde = "0.1"
proxmox-shared-memory = "0.2"
proxmox-sys = "0.2"
+proxmox-metrics = "0.1"
pbs-api-types = { path = "../pbs-api-types" }
pbs-buildcfg = { path = "../pbs-buildcfg" }
diff --git a/pbs-config/src/lib.rs b/pbs-config/src/lib.rs
index 118030bc..29880ab9 100644
--- a/pbs-config/src/lib.rs
+++ b/pbs-config/src/lib.rs
@@ -6,6 +6,7 @@ pub mod domains;
pub mod drive;
pub mod key_config;
pub mod media_pool;
+pub mod metrics;
pub mod network;
pub mod remote;
pub mod sync;
diff --git a/pbs-config/src/metrics.rs b/pbs-config/src/metrics.rs
new file mode 100644
index 00000000..39096ac2
--- /dev/null
+++ b/pbs-config/src/metrics.rs
@@ -0,0 +1,128 @@
+use std::collections::HashMap;
+
+use anyhow::Error;
+use lazy_static::lazy_static;
+
+use proxmox_metrics::{influxdb, MetricsChannel, MetricsServerFuture};
+use proxmox_schema::*;
+use proxmox_section_config::{SectionConfig, SectionConfigData, SectionConfigPlugin};
+
+use pbs_api_types::{InfluxDbHttp, InfluxDbUdp, METRIC_SERVER_ID_SCHEMA};
+
+use crate::{open_backup_lockfile, BackupLockGuard};
+
+lazy_static! {
+ pub static ref CONFIG: SectionConfig = init();
+}
+
+fn init() -> SectionConfig {
+ let mut config = SectionConfig::new(&METRIC_SERVER_ID_SCHEMA);
+
+ let udp_schema = match InfluxDbUdp::API_SCHEMA {
+ Schema::Object(ref object_schema) => object_schema,
+ _ => unreachable!(),
+ };
+
+ let udp_plugin = SectionConfigPlugin::new(
+ "influxdb-udp".to_string(),
+ Some("name".to_string()),
+ udp_schema,
+ );
+ config.register_plugin(udp_plugin);
+
+ let http_schema = match InfluxDbHttp::API_SCHEMA {
+ Schema::Object(ref object_schema) => object_schema,
+ _ => unreachable!(),
+ };
+
+ let http_plugin = SectionConfigPlugin::new(
+ "influxdb-http".to_string(),
+ Some("name".to_string()),
+ http_schema,
+ );
+
+ config.register_plugin(http_plugin);
+
+ config
+}
+
+pub const METRIC_SERVER_CFG_FILENAME: &str = "/etc/proxmox-backup/metricserver.cfg";
+pub const METRIC_SERVER_CFG_LOCKFILE: &str = "/etc/proxmox-backup/.metricserver.lck";
+
+/// Get exclusive lock
+pub fn lock_config() -> Result<BackupLockGuard, Error> {
+ open_backup_lockfile(METRIC_SERVER_CFG_LOCKFILE, None, true)
+}
+
+pub fn config() -> Result<(SectionConfigData, [u8; 32]), Error> {
+ let content = proxmox_sys::fs::file_read_optional_string(METRIC_SERVER_CFG_FILENAME)?
+ .unwrap_or_else(|| "".to_string());
+
+ let digest = openssl::sha::sha256(content.as_bytes());
+ let data = CONFIG.parse(METRIC_SERVER_CFG_FILENAME, &content)?;
+ Ok((data, digest))
+}
+
+pub fn save_config(config: &SectionConfigData) -> Result<(), Error> {
+ let raw = CONFIG.write(METRIC_SERVER_CFG_FILENAME, config)?;
+ crate::replace_backup_config(METRIC_SERVER_CFG_FILENAME, raw.as_bytes())
+}
+
+// shell completion helper
+pub fn complete_remote_name(_arg: &str, _param: &HashMap<String, String>) -> Vec<String> {
+ match config() {
+ Ok((data, _digest)) => data.sections.iter().map(|(id, _)| id.to_string()).collect(),
+ Err(_) => return vec![],
+ }
+}
+
+/// Get the metric server connections from a config
+pub fn get_metric_server_connections(
+ metric_config: SectionConfigData,
+) -> Result<(Vec<MetricsServerFuture>, Vec<MetricsChannel>, Vec<String>), Error> {
+ let mut futures = Vec::new();
+ let mut channels = Vec::new();
+ let mut names = Vec::new();
+
+ for config in
+ metric_config.convert_to_typed_array::<pbs_api_types::InfluxDbUdp>("influxdb-udp")?
+ {
+ if !config.enable.unwrap_or(true) {
+ continue;
+ }
+ let (future, sender) = influxdb::InfluxDbUdp::new(&config.host, config.port, config.mtu);
+ names.push(config.name);
+ futures.push(future);
+ channels.push(sender);
+ }
+
+ for config in
+ metric_config.convert_to_typed_array::<pbs_api_types::InfluxDbHttp>("influxdb-http")?
+ {
+ if !config.enable.unwrap_or(true) {
+ continue;
+ }
+ let port = config.port.unwrap_or_else(|| {
+ if config.https.unwrap_or(true) {
+ 443
+ } else {
+ 80
+ }
+ });
+
+ let (future, sender) = influxdb::InfluxDbHttp::new(
+ config.https.unwrap_or(true),
+ &config.host,
+ port,
+ config.organization.as_deref().unwrap_or("proxmox"),
+ config.bucket.as_deref().unwrap_or("proxmox"),
+ config.token.as_deref(),
+ config.verify_tls.unwrap_or(true),
+ config.max_body_size.unwrap_or(25_000_000),
+ )?;
+ names.push(config.name);
+ futures.push(future);
+ channels.push(sender);
+ }
+ Ok((futures, channels, names))
+}
--
2.30.2
^ permalink raw reply [flat|nested] 13+ messages in thread
* [pbs-devel] [PATCH proxmox-backup v3 4/6] backup-proxy: decouple stats gathering from rrd update
2021-12-17 8:09 [pbs-devel] [PATCH proxmox/proxmox-backup v3] add metrics server capability Dominik Csapak
` (5 preceding siblings ...)
2021-12-17 8:09 ` [pbs-devel] [PATCH proxmox-backup v3 3/6] pbs-config: add metrics config class Dominik Csapak
@ 2021-12-17 8:09 ` Dominik Csapak
2021-12-17 8:09 ` [pbs-devel] [PATCH proxmox-backup v3 5/6] proxmox-backup-proxy: send metrics to configured metrics server Dominik Csapak
2021-12-17 8:10 ` [pbs-devel] [PATCH proxmox-backup v3 6/6] api: add metricserver endpoints Dominik Csapak
8 siblings, 0 replies; 13+ messages in thread
From: Dominik Csapak @ 2021-12-17 8:09 UTC (permalink / raw)
To: pbs-devel
that way we can reuse the stats gathered
Signed-off-by: Dominik Csapak <d.csapak@proxmox.com>
---
src/bin/proxmox-backup-proxy.rs | 213 +++++++++++++++++++++-----------
1 file changed, 141 insertions(+), 72 deletions(-)
diff --git a/src/bin/proxmox-backup-proxy.rs b/src/bin/proxmox-backup-proxy.rs
index fbc5f65c..34907581 100644
--- a/src/bin/proxmox-backup-proxy.rs
+++ b/src/bin/proxmox-backup-proxy.rs
@@ -17,8 +17,11 @@ use tokio_stream::wrappers::ReceiverStream;
use serde_json::{json, Value};
use http::{Method, HeaderMap};
-use proxmox_sys::linux::socket::set_tcp_keepalive;
-use proxmox_sys::fs::CreateOptions;
+use proxmox_sys::linux::{
+ procfs::{ProcFsStat, ProcFsMemInfo, ProcFsNetDev, Loadavg},
+ socket::set_tcp_keepalive
+};
+use proxmox_sys::fs::{CreateOptions, FileSystemInformation};
use proxmox_lang::try_block;
use proxmox_router::{RpcEnvironment, RpcEnvironmentType, UserInformation};
use proxmox_http::client::{RateLimitedStream, ShareableRateLimit};
@@ -44,6 +47,7 @@ use proxmox_backup::{
Job,
},
},
+ tools::disks::BlockDevStat,
};
use pbs_buildcfg::configdir;
@@ -931,9 +935,24 @@ async fn run_stat_generator() {
loop {
let delay_target = Instant::now() + Duration::from_secs(10);
- generate_host_stats().await;
+ let stats = match tokio::task::spawn_blocking(|| {
+ let hoststats = collect_host_stats_sync();
+ let (hostdisk, datastores) = collect_disk_stats_sync();
+ Arc::new((hoststats, hostdisk, datastores))
+ }).await {
+ Ok(res) => res,
+ Err(err) => {
+ log::error!("collecting host stats paniced: {}", err);
+ tokio::time::sleep_until(tokio::time::Instant::from_std(delay_target)).await;
+ continue;
+ }
+ };
+
+ let rrd_future = tokio::task::spawn_blocking(move || {
+ rrd_update_host_stats_sync(&stats.0, &stats.1, &stats.2);
+ rrd_sync_journal();
+ });
- rrd_sync_journal();
tokio::time::sleep_until(tokio::time::Instant::from_std(delay_target)).await;
@@ -941,86 +960,147 @@ async fn run_stat_generator() {
}
-async fn generate_host_stats() {
- match tokio::task::spawn_blocking(generate_host_stats_sync).await {
- Ok(()) => (),
- Err(err) => log::error!("generate_host_stats paniced: {}", err),
- }
+struct HostStats {
+ proc: Option<ProcFsStat>,
+ meminfo: Option<ProcFsMemInfo>,
+ net: Option<Vec<ProcFsNetDev>>,
+ load: Option<Loadavg>,
+}
+
+struct DiskStat {
+ name: String,
+ usage: Option<FileSystemInformation>,
+ dev: Option<BlockDevStat>,
}
-fn generate_host_stats_sync() {
+fn collect_host_stats_sync() -> HostStats {
use proxmox_sys::linux::procfs::{
read_meminfo, read_proc_stat, read_proc_net_dev, read_loadavg};
- match read_proc_stat() {
- Ok(stat) => {
- rrd_update_gauge("host/cpu", stat.cpu);
- rrd_update_gauge("host/iowait", stat.iowait_percent);
- }
+ let proc = match read_proc_stat() {
+ Ok(stat) => Some(stat),
Err(err) => {
eprintln!("read_proc_stat failed - {}", err);
+ None
}
- }
+ };
- match read_meminfo() {
- Ok(meminfo) => {
- rrd_update_gauge("host/memtotal", meminfo.memtotal as f64);
- rrd_update_gauge("host/memused", meminfo.memused as f64);
- rrd_update_gauge("host/swaptotal", meminfo.swaptotal as f64);
- rrd_update_gauge("host/swapused", meminfo.swapused as f64);
- }
+ let meminfo = match read_meminfo() {
+ Ok(stat) => Some(stat),
Err(err) => {
eprintln!("read_meminfo failed - {}", err);
+ None
}
- }
+ };
- match read_proc_net_dev() {
- Ok(netdev) => {
- use pbs_config::network::is_physical_nic;
- let mut netin = 0;
- let mut netout = 0;
- for item in netdev {
- if !is_physical_nic(&item.device) { continue; }
- netin += item.receive;
- netout += item.send;
- }
- rrd_update_derive("host/netin", netin as f64);
- rrd_update_derive("host/netout", netout as f64);
- }
+ let net = match read_proc_net_dev() {
+ Ok(netdev) => Some(netdev),
Err(err) => {
eprintln!("read_prox_net_dev failed - {}", err);
+ None
}
- }
+ };
- match read_loadavg() {
- Ok(loadavg) => {
- rrd_update_gauge("host/loadavg", loadavg.0 as f64);
- }
+ let load = match read_loadavg() {
+ Ok(loadavg) => Some(loadavg),
Err(err) => {
eprintln!("read_loadavg failed - {}", err);
+ None
}
+ };
+
+ HostStats {
+ proc,
+ meminfo,
+ net,
+ load,
}
+}
+fn collect_disk_stats_sync() -> (DiskStat, Vec<DiskStat>) {
let disk_manager = DiskManage::new();
- gather_disk_stats(disk_manager.clone(), Path::new("/"), "host");
+ let root = gather_disk_stats(disk_manager.clone(), Path::new("/"), "host");
+ let mut datastores = Vec::new();
match pbs_config::datastore::config() {
Ok((config, _)) => {
let datastore_list: Vec<DataStoreConfig> =
config.convert_to_typed_array("datastore").unwrap_or_default();
for config in datastore_list {
-
- let rrd_prefix = format!("datastore/{}", config.name);
let path = std::path::Path::new(&config.path);
- gather_disk_stats(disk_manager.clone(), path, &rrd_prefix);
+ datastores.push(gather_disk_stats(disk_manager.clone(), path, &config.name));
}
}
Err(err) => {
eprintln!("read datastore config failed - {}", err);
}
}
+
+ (root, datastores)
+}
+
+fn rrd_update_host_stats_sync(host: &HostStats, hostdisk: &DiskStat, datastores: &[DiskStat]) {
+ if let Some(stat) = &host.proc {
+ rrd_update_gauge("host/cpu", stat.cpu);
+ rrd_update_gauge("host/iowait", stat.iowait_percent);
+ }
+
+ if let Some(meminfo) = &host.meminfo {
+ rrd_update_gauge("host/memtotal", meminfo.memtotal as f64);
+ rrd_update_gauge("host/memused", meminfo.memused as f64);
+ rrd_update_gauge("host/swaptotal", meminfo.swaptotal as f64);
+ rrd_update_gauge("host/swapused", meminfo.swapused as f64);
+ }
+
+ if let Some(netdev) = &host.net {
+ use pbs_config::network::is_physical_nic;
+ let mut netin = 0;
+ let mut netout = 0;
+ for item in netdev {
+ if !is_physical_nic(&item.device) { continue; }
+ netin += item.receive;
+ netout += item.send;
+ }
+ rrd_update_derive("host/netin", netin as f64);
+ rrd_update_derive("host/netout", netout as f64);
+ }
+
+ if let Some(loadavg) = &host.load {
+ rrd_update_gauge("host/loadavg", loadavg.0 as f64);
+ }
+
+ rrd_update_disk_stat(&hostdisk, "host");
+
+ for stat in datastores {
+ let rrd_prefix = format!("datastore/{}", stat.name);
+ rrd_update_disk_stat(&stat, &rrd_prefix);
+ }
+}
+
+fn rrd_update_disk_stat(disk: &DiskStat, rrd_prefix: &str) {
+ if let Some(status) = &disk.usage {
+ let rrd_key = format!("{}/total", rrd_prefix);
+ rrd_update_gauge(&rrd_key, status.total as f64);
+ let rrd_key = format!("{}/used", rrd_prefix);
+ rrd_update_gauge(&rrd_key, status.used as f64);
+ }
+
+ if let Some(stat) = &disk.dev {
+ let rrd_key = format!("{}/read_ios", rrd_prefix);
+ rrd_update_derive(&rrd_key, stat.read_ios as f64);
+ let rrd_key = format!("{}/read_bytes", rrd_prefix);
+ rrd_update_derive(&rrd_key, (stat.read_sectors*512) as f64);
+
+ let rrd_key = format!("{}/write_ios", rrd_prefix);
+ rrd_update_derive(&rrd_key, stat.write_ios as f64);
+ let rrd_key = format!("{}/write_bytes", rrd_prefix);
+ rrd_update_derive(&rrd_key, (stat.write_sectors*512) as f64);
+
+ let rrd_key = format!("{}/io_ticks", rrd_prefix);
+ rrd_update_derive(&rrd_key, (stat.io_ticks as f64)/1000.0);
+ }
}
fn check_schedule(worker_type: &str, event_str: &str, id: &str) -> bool {
@@ -1053,22 +1133,17 @@ fn check_schedule(worker_type: &str, event_str: &str, id: &str) -> bool {
next <= now
}
-fn gather_disk_stats(disk_manager: Arc<DiskManage>, path: &Path, rrd_prefix: &str) {
-
- match proxmox_sys::fs::fs_info(path) {
- Ok(status) => {
- let rrd_key = format!("{}/total", rrd_prefix);
- rrd_update_gauge(&rrd_key, status.total as f64);
- let rrd_key = format!("{}/used", rrd_prefix);
- rrd_update_gauge(&rrd_key, status.used as f64);
- }
+fn gather_disk_stats(disk_manager: Arc<DiskManage>, path: &Path, name: &str) -> DiskStat {
+ let usage = match proxmox_sys::fs::fs_info(path) {
+ Ok(status) => Some(status),
Err(err) => {
eprintln!("read fs info on {:?} failed - {}", path, err);
+ None
}
- }
+ };
- match disk_manager.find_mounted_device(path) {
- Ok(None) => {},
+ let dev = match disk_manager.find_mounted_device(path) {
+ Ok(None) => None,
Ok(Some((fs_type, device, source))) => {
let mut device_stat = None;
match fs_type.as_str() {
@@ -1090,24 +1165,18 @@ fn gather_disk_stats(disk_manager: Arc<DiskManage>, path: &Path, rrd_prefix: &st
}
}
}
- if let Some(stat) = device_stat {
- let rrd_key = format!("{}/read_ios", rrd_prefix);
- rrd_update_derive(&rrd_key, stat.read_ios as f64);
- let rrd_key = format!("{}/read_bytes", rrd_prefix);
- rrd_update_derive(&rrd_key, (stat.read_sectors*512) as f64);
-
- let rrd_key = format!("{}/write_ios", rrd_prefix);
- rrd_update_derive(&rrd_key, stat.write_ios as f64);
- let rrd_key = format!("{}/write_bytes", rrd_prefix);
- rrd_update_derive(&rrd_key, (stat.write_sectors*512) as f64);
-
- let rrd_key = format!("{}/io_ticks", rrd_prefix);
- rrd_update_derive(&rrd_key, (stat.io_ticks as f64)/1000.0);
- }
+ device_stat
}
Err(err) => {
eprintln!("find_mounted_device failed - {}", err);
+ None
}
+ };
+
+ DiskStat {
+ name: name.to_string(),
+ usage,
+ dev,
}
}
--
2.30.2
^ permalink raw reply [flat|nested] 13+ messages in thread
* [pbs-devel] [PATCH proxmox-backup v3 5/6] proxmox-backup-proxy: send metrics to configured metrics server
2021-12-17 8:09 [pbs-devel] [PATCH proxmox/proxmox-backup v3] add metrics server capability Dominik Csapak
` (6 preceding siblings ...)
2021-12-17 8:09 ` [pbs-devel] [PATCH proxmox-backup v3 4/6] backup-proxy: decouple stats gathering from rrd update Dominik Csapak
@ 2021-12-17 8:09 ` Dominik Csapak
2021-12-17 8:10 ` [pbs-devel] [PATCH proxmox-backup v3 6/6] api: add metricserver endpoints Dominik Csapak
8 siblings, 0 replies; 13+ messages in thread
From: Dominik Csapak @ 2021-12-17 8:09 UTC (permalink / raw)
To: pbs-devel
and keep the data as similar as possible to pve (tags/fields)
datastores get their own 'object' type and reside in the "blockstat"
measurement
Signed-off-by: Dominik Csapak <d.csapak@proxmox.com>
---
src/bin/proxmox-backup-proxy.rs | 139 +++++++++++++++++++++++++++++++-
1 file changed, 136 insertions(+), 3 deletions(-)
diff --git a/src/bin/proxmox-backup-proxy.rs b/src/bin/proxmox-backup-proxy.rs
index 34907581..0638936b 100644
--- a/src/bin/proxmox-backup-proxy.rs
+++ b/src/bin/proxmox-backup-proxy.rs
@@ -23,11 +23,13 @@ use proxmox_sys::linux::{
};
use proxmox_sys::fs::{CreateOptions, FileSystemInformation};
use proxmox_lang::try_block;
+use proxmox_metrics::MetricsData;
use proxmox_router::{RpcEnvironment, RpcEnvironmentType, UserInformation};
use proxmox_http::client::{RateLimitedStream, ShareableRateLimit};
use proxmox_sys::{task_log, task_warn};
use proxmox_sys::logrotate::LogRotate;
+use pbs_config::metrics::get_metric_server_connections;
use pbs_datastore::DataStore;
use proxmox_rest_server::{
@@ -948,16 +950,127 @@ async fn run_stat_generator() {
}
};
- let rrd_future = tokio::task::spawn_blocking(move || {
- rrd_update_host_stats_sync(&stats.0, &stats.1, &stats.2);
- rrd_sync_journal();
+ let rrd_future = tokio::task::spawn_blocking({
+ let stats = Arc::clone(&stats);
+ move || {
+ rrd_update_host_stats_sync(&stats.0, &stats.1, &stats.2);
+ rrd_sync_journal();
+ }
});
+ let metrics_future = send_data_to_metric_servers(stats);
+
+ let (rrd_res, metrics_res) = join!(rrd_future, metrics_future);
+ if let Err(err) = rrd_res {
+ log::error!("rrd update panicked: {}", err);
+ }
+ if let Err(err) = metrics_res {
+ log::error!("error during metrics sending: {}", err);
+ }
tokio::time::sleep_until(tokio::time::Instant::from_std(delay_target)).await;
}
+}
+
+async fn send_data_to_metric_servers(
+ stats: Arc<(HostStats, DiskStat, Vec<DiskStat>)>,
+) -> Result<(), Error> {
+ let (config, _digest) = pbs_config::metrics::config()?;
+ let (futures, channels, names) = get_metric_server_connections(config)?;
+
+ if futures.is_empty() {
+ return Ok(());
+ }
+
+ let sending_handle = tokio::spawn(async move {
+ future::join_all(futures).await
+ });
+
+ let ctime = proxmox_time::epoch_i64();
+ let nodename = proxmox_sys::nodename();
+
+ let mut values = Vec::new();
+
+ let mut cpuvalue = match &stats.0.proc {
+ Some(stat) => serde_json::to_value(stat)?,
+ None => json!({}),
+ };
+
+ if let Some(loadavg) = &stats.0.load {
+ cpuvalue["avg1"] = Value::from(loadavg.0);
+ cpuvalue["avg5"] = Value::from(loadavg.1);
+ cpuvalue["avg15"] = Value::from(loadavg.2);
+ }
+
+ values.push(Arc::new(MetricsData::new(
+ "cpustat",
+ &[("object", "host"), ("host", nodename)],
+ ctime,
+ cpuvalue,
+ )?));
+
+ if let Some(stat) = &stats.0.meminfo {
+ values.push(Arc::new(MetricsData::new(
+ "memory",
+ &[("object", "host"), ("host", nodename)],
+ ctime,
+ stat,
+ )?));
+ }
+ if let Some(netdev) = &stats.0.net {
+ for item in netdev {
+ values.push(Arc::new(MetricsData::new(
+ "nics",
+ &[
+ ("object", "host"),
+ ("host", nodename),
+ ("instance", &item.device),
+ ],
+ ctime,
+ item,
+ )?));
+ }
+ }
+
+ values.push(Arc::new(MetricsData::new(
+ "blockstat",
+ &[("object", "host"), ("host", nodename)],
+ ctime,
+ stats.1.to_value(),
+ )?));
+
+ for datastore in stats.2.iter() {
+ values.push(Arc::new(MetricsData::new(
+ "blockstat",
+ &[
+ ("object", "datastore"),
+ ("nodename", nodename),
+ ("datastore", &datastore.name),
+ ],
+ ctime,
+ datastore.to_value(),
+ )?));
+ }
+
+ let results = proxmox_metrics::send_data_to_channels(&values, &channels).await;
+ for (i, res) in results.into_iter().enumerate() {
+ if let Err(err) = res {
+ log::error!("error sending into channel of {}: {}", names[i], err);
+ }
+ }
+
+ drop(channels);
+
+ let results = sending_handle.await?;
+ for (i, res) in results.into_iter().enumerate() {
+ if let Err(err) = res {
+ log::error!("error sending to metric server {}: {}", names[i], err);
+ }
+ }
+
+ Ok(())
}
struct HostStats {
@@ -973,6 +1086,26 @@ struct DiskStat {
dev: Option<BlockDevStat>,
}
+impl DiskStat {
+ fn to_value(&self) -> Value {
+ let mut value = json!({});
+ if let Some(usage) = &self.usage {
+ value["total"] = Value::from(usage.total);
+ value["used"] = Value::from(usage.used);
+ value["avail"] = Value::from(usage.available);
+ }
+
+ if let Some(dev) = &self.dev {
+ value["read_ios"] = Value::from(dev.read_ios);
+ value["read_bytes"] = Value::from(dev.read_sectors * 512);
+ value["write_ios"] = Value::from(dev.write_ios);
+ value["write_bytes"] = Value::from(dev.write_sectors * 512);
+ value["io_ticks"] = Value::from(dev.io_ticks / 1000);
+ }
+ value
+ }
+}
+
fn collect_host_stats_sync() -> HostStats {
use proxmox_sys::linux::procfs::{
read_meminfo, read_proc_stat, read_proc_net_dev, read_loadavg};
--
2.30.2
^ permalink raw reply [flat|nested] 13+ messages in thread
* [pbs-devel] [PATCH proxmox-backup v3 6/6] api: add metricserver endpoints
2021-12-17 8:09 [pbs-devel] [PATCH proxmox/proxmox-backup v3] add metrics server capability Dominik Csapak
` (7 preceding siblings ...)
2021-12-17 8:09 ` [pbs-devel] [PATCH proxmox-backup v3 5/6] proxmox-backup-proxy: send metrics to configured metrics server Dominik Csapak
@ 2021-12-17 8:10 ` Dominik Csapak
8 siblings, 0 replies; 13+ messages in thread
From: Dominik Csapak @ 2021-12-17 8:10 UTC (permalink / raw)
To: pbs-devel
but in contrast to pve, we split the api by type of the section config,
since we cannot handle multiple types in the updater
Signed-off-by: Dominik Csapak <d.csapak@proxmox.com>
---
src/api2/config/metricserver/influxdbhttp.rs | 275 +++++++++++++++++++
src/api2/config/metricserver/influxdbudp.rs | 245 +++++++++++++++++
src/api2/config/metricserver/mod.rs | 16 ++
src/api2/config/mod.rs | 2 +
4 files changed, 538 insertions(+)
create mode 100644 src/api2/config/metricserver/influxdbhttp.rs
create mode 100644 src/api2/config/metricserver/influxdbudp.rs
create mode 100644 src/api2/config/metricserver/mod.rs
diff --git a/src/api2/config/metricserver/influxdbhttp.rs b/src/api2/config/metricserver/influxdbhttp.rs
new file mode 100644
index 00000000..aea3f7d2
--- /dev/null
+++ b/src/api2/config/metricserver/influxdbhttp.rs
@@ -0,0 +1,275 @@
+use anyhow::{bail, Error};
+use serde_json::Value;
+use serde::{Deserialize, Serialize};
+use hex::FromHex;
+
+use proxmox_router::{Router, RpcEnvironment, Permission};
+use proxmox_schema::api;
+
+use pbs_api_types::{
+ InfluxDbHttp, InfluxDbHttpUpdater,
+ PROXMOX_CONFIG_DIGEST_SCHEMA, METRIC_SERVER_ID_SCHEMA, PRIV_SYS_AUDIT, PRIV_SYS_MODIFY,
+};
+
+use pbs_config::metrics;
+
+#[api(
+ input: {
+ properties: {},
+ },
+ returns: {
+ description: "List of configured InfluxDB http metric servers.",
+ type: Array,
+ items: { type: InfluxDbHttp },
+ },
+ access: {
+ permission: &Permission::Privilege(&[], PRIV_SYS_AUDIT, false),
+ },
+)]
+/// List configured InfluxDB http metric servers.
+pub fn list_influxdb_http_servers(
+ _param: Value,
+ mut rpcenv: &mut dyn RpcEnvironment,
+) -> Result<Vec<InfluxDbHttp>, Error> {
+
+ let (config, digest) = metrics::config()?;
+
+ let mut list: Vec<InfluxDbHttp> = config.convert_to_typed_array("influxdb-http")?;
+
+ // don't return token via api
+ for item in list.iter_mut() {
+ item.token = None;
+ }
+
+ rpcenv["digest"] = hex::encode(&digest).into();
+
+ Ok(list)
+}
+
+#[api(
+ protected: true,
+ input: {
+ properties: {
+ config: {
+ type: InfluxDbHttp,
+ flatten: true,
+ },
+ },
+ },
+ access: {
+ permission: &Permission::Privilege(&[], PRIV_SYS_MODIFY, false),
+ },
+)]
+/// Create a new InfluxDB http server configuration
+pub fn create_influxdb_http_server(config: InfluxDbHttp) -> Result<(), Error> {
+
+ let _lock = metrics::lock_config()?;
+
+ let (mut metrics, _digest) = metrics::config()?;
+
+ metrics.set_data(&config.name, "influxdb-http", &config)?;
+
+ metrics::save_config(&metrics)?;
+
+ Ok(())
+}
+
+#[api(
+ protected: true,
+ input: {
+ properties: {
+ name: {
+ schema: METRIC_SERVER_ID_SCHEMA,
+ },
+ digest: {
+ optional: true,
+ schema: PROXMOX_CONFIG_DIGEST_SCHEMA,
+ },
+ },
+ },
+ access: {
+ permission: &Permission::Privilege(&[], PRIV_SYS_MODIFY, false),
+ },
+)]
+/// Remove a InfluxDB http server configuration
+pub fn delete_influxdb_http_server(
+ name: String,
+ digest: Option<String>,
+ _rpcenv: &mut dyn RpcEnvironment,
+) -> Result<(), Error> {
+
+ let _lock = metrics::lock_config()?;
+
+ let (mut metrics, expected_digest) = metrics::config()?;
+
+ if let Some(ref digest) = digest {
+ let digest = <[u8; 32]>::from_hex(digest)?;
+ crate::tools::detect_modified_configuration_file(&digest, &expected_digest)?;
+ }
+
+ if metrics.sections.remove(&name).is_none() {
+ bail!("name '{}' does not exist.", name);
+ }
+
+ metrics::save_config(&metrics)?;
+
+ Ok(())
+}
+
+#[api(
+ input: {
+ properties: {
+ name: {
+ schema: METRIC_SERVER_ID_SCHEMA,
+ },
+ },
+ },
+ returns: { type: InfluxDbHttp },
+ access: {
+ permission: &Permission::Privilege(&[], PRIV_SYS_AUDIT, false),
+ },
+)]
+/// Read the InfluxDB http server configuration
+pub fn read_influxdb_http_server(
+ name: String,
+ mut rpcenv: &mut dyn RpcEnvironment,
+) -> Result<InfluxDbHttp, Error> {
+
+ let (metrics, digest) = metrics::config()?;
+
+ let mut config: InfluxDbHttp = metrics.lookup("influxdb-http", &name)?;
+
+ config.token = None;
+
+ rpcenv["digest"] = hex::encode(&digest).into();
+
+ Ok(config)
+}
+
+#[api()]
+#[derive(Serialize, Deserialize)]
+#[serde(rename_all="kebab-case")]
+/// Deletable property name
+pub enum DeletableProperty {
+ /// Delete the enable property.
+ Enable,
+ /// Delete the port property.
+ Port,
+ /// Delete the https property.
+ Https,
+ /// Delete the token property.
+ Token,
+ /// Delete the bucket property.
+ Bucket,
+ /// Delete the organization property.
+ Organization,
+ /// Delete the max_body_size property.
+ MaxBodySize,
+ /// Delete the verify_tls property.
+ VerifyTls,
+ /// Delete the comment property.
+ Comment,
+}
+
+#[api(
+ protected: true,
+ input: {
+ properties: {
+ name: {
+ schema: METRIC_SERVER_ID_SCHEMA,
+ },
+ update: {
+ type: InfluxDbHttpUpdater,
+ flatten: true,
+ },
+ delete: {
+ description: "List of properties to delete.",
+ type: Array,
+ optional: true,
+ items: {
+ type: DeletableProperty,
+ }
+ },
+ digest: {
+ optional: true,
+ schema: PROXMOX_CONFIG_DIGEST_SCHEMA,
+ },
+ },
+ },
+ returns: { type: InfluxDbHttp },
+ access: {
+ permission: &Permission::Privilege(&[], PRIV_SYS_MODIFY, false),
+ },
+)]
+/// Update an InfluxDB http server configuration
+pub fn update_influxdb_http_server(
+ name: String,
+ update: InfluxDbHttpUpdater,
+ delete: Option<Vec<DeletableProperty>>,
+ digest: Option<String>,
+ _rpcenv: &mut dyn RpcEnvironment,
+) -> Result<(), Error> {
+
+ let _lock = metrics::lock_config()?;
+
+ let (mut metrics, expected_digest) = metrics::config()?;
+
+ if let Some(ref digest) = digest {
+ let digest = <[u8; 32]>::from_hex(digest)?;
+ crate::tools::detect_modified_configuration_file(&digest, &expected_digest)?;
+ }
+
+ let mut config: InfluxDbHttp = metrics.lookup("influxdb-http", &name)?;
+
+ if let Some(delete) = delete {
+ for delete_prop in delete {
+ match delete_prop {
+ DeletableProperty::Enable => { config.enable = None; },
+ DeletableProperty::Port => { config.port = None; },
+ DeletableProperty::Https => { config.https = None; },
+ DeletableProperty::Token => { config.token = None; },
+ DeletableProperty::Bucket => { config.bucket = None; },
+ DeletableProperty::Organization => { config.organization = None; },
+ DeletableProperty::MaxBodySize => { config.max_body_size = None; },
+ DeletableProperty::VerifyTls => { config.verify_tls = None; },
+ DeletableProperty::Comment => { config.comment = None; },
+ }
+ }
+ }
+
+ if let Some(comment) = update.comment {
+ let comment = comment.trim().to_string();
+ if comment.is_empty() {
+ config.comment = None;
+ } else {
+ config.comment = Some(comment);
+ }
+ }
+
+ if let Some(host) = update.host { config.host = host; }
+
+ if update.enable.is_some() { config.enable = update.enable; }
+ if update.port.is_some() { config.port = update.port; }
+ if update.https.is_some() { config.https = update.https; }
+ if update.token.is_some() { config.token = update.token; }
+ if update.bucket.is_some() { config.bucket = update.bucket; }
+ if update.organization.is_some() { config.organization = update.organization; }
+ if update.max_body_size.is_some() { config.max_body_size = update.max_body_size; }
+ if update.verify_tls.is_some() { config.verify_tls = update.verify_tls; }
+
+ metrics.set_data(&name, "influxdb-http", &config)?;
+
+ metrics::save_config(&metrics)?;
+
+ Ok(())
+}
+
+const ITEM_ROUTER: Router = Router::new()
+ .get(&API_METHOD_READ_INFLUXDB_HTTP_SERVER)
+ .put(&API_METHOD_UPDATE_INFLUXDB_HTTP_SERVER)
+ .delete(&API_METHOD_DELETE_INFLUXDB_HTTP_SERVER);
+
+pub const ROUTER: Router = Router::new()
+ .get(&API_METHOD_LIST_INFLUXDB_HTTP_SERVERS)
+ .post(&API_METHOD_CREATE_INFLUXDB_HTTP_SERVER)
+ .match_all("name", &ITEM_ROUTER);
diff --git a/src/api2/config/metricserver/influxdbudp.rs b/src/api2/config/metricserver/influxdbudp.rs
new file mode 100644
index 00000000..2cb91422
--- /dev/null
+++ b/src/api2/config/metricserver/influxdbudp.rs
@@ -0,0 +1,245 @@
+use anyhow::{bail, Error};
+use serde_json::Value;
+use serde::{Deserialize, Serialize};
+use hex::FromHex;
+
+use proxmox_router::{Router, RpcEnvironment, Permission};
+use proxmox_schema::api;
+
+use pbs_api_types::{
+ InfluxDbUdp, InfluxDbUdpUpdater,
+ PROXMOX_CONFIG_DIGEST_SCHEMA, METRIC_SERVER_ID_SCHEMA, PRIV_SYS_AUDIT, PRIV_SYS_MODIFY,
+};
+
+use pbs_config::metrics;
+
+#[api(
+ input: {
+ properties: {},
+ },
+ returns: {
+ description: "List of configured InfluxDB udp metric servers.",
+ type: Array,
+ items: { type: InfluxDbUdp },
+ },
+ access: {
+ permission: &Permission::Privilege(&[], PRIV_SYS_AUDIT, false),
+ },
+)]
+/// List configured InfluxDB udp metric servers.
+pub fn list_influxdb_udp_servers(
+ _param: Value,
+ mut rpcenv: &mut dyn RpcEnvironment,
+) -> Result<Vec<InfluxDbUdp>, Error> {
+
+ let (config, digest) = metrics::config()?;
+
+ let list = config.convert_to_typed_array("influxdb-udp")?;
+
+ rpcenv["digest"] = hex::encode(&digest).into();
+
+ Ok(list)
+}
+
+#[api(
+ protected: true,
+ input: {
+ properties: {
+ config: {
+ type: InfluxDbUdp,
+ flatten: true,
+ },
+ },
+ },
+ access: {
+ permission: &Permission::Privilege(&[], PRIV_SYS_MODIFY, false),
+ },
+)]
+/// Create a new InfluxDB udp server configuration
+pub fn create_influxdb_udp_server(config: InfluxDbUdp) -> Result<(), Error> {
+
+ let _lock = metrics::lock_config()?;
+
+ let (mut metrics, _digest) = metrics::config()?;
+
+ metrics.set_data(&config.name, "influxdb-udp", &config)?;
+
+ metrics::save_config(&metrics)?;
+
+ Ok(())
+}
+
+#[api(
+ protected: true,
+ input: {
+ properties: {
+ name: {
+ schema: METRIC_SERVER_ID_SCHEMA,
+ },
+ digest: {
+ optional: true,
+ schema: PROXMOX_CONFIG_DIGEST_SCHEMA,
+ },
+ },
+ },
+ access: {
+ permission: &Permission::Privilege(&[], PRIV_SYS_MODIFY, false),
+ },
+)]
+/// Remove a InfluxDB udp server configuration
+pub fn delete_influxdb_udp_server(
+ name: String,
+ digest: Option<String>,
+ _rpcenv: &mut dyn RpcEnvironment,
+) -> Result<(), Error> {
+
+ let _lock = metrics::lock_config()?;
+
+ let (mut metrics, expected_digest) = metrics::config()?;
+
+ if let Some(ref digest) = digest {
+ let digest = <[u8; 32]>::from_hex(digest)?;
+ crate::tools::detect_modified_configuration_file(&digest, &expected_digest)?;
+ }
+
+ if metrics.sections.remove(&name).is_none() {
+ bail!("name '{}' does not exist.", name);
+ }
+
+ metrics::save_config(&metrics)?;
+
+ Ok(())
+}
+
+#[api(
+ input: {
+ properties: {
+ name: {
+ schema: METRIC_SERVER_ID_SCHEMA,
+ },
+ },
+ },
+ returns: { type: InfluxDbUdp },
+ access: {
+ permission: &Permission::Privilege(&[], PRIV_SYS_AUDIT, false),
+ },
+)]
+/// Read the InfluxDB udp server configuration
+pub fn read_influxdb_udp_server(
+ name: String,
+ mut rpcenv: &mut dyn RpcEnvironment,
+) -> Result<InfluxDbUdp, Error> {
+
+ let (metrics, digest) = metrics::config()?;
+
+ let config = metrics.lookup("influxdb-udp", &name)?;
+
+ rpcenv["digest"] = hex::encode(&digest).into();
+
+ Ok(config)
+}
+
+#[api()]
+#[derive(Serialize, Deserialize)]
+#[serde(rename_all="kebab-case")]
+/// Deletable property name
+pub enum DeletableProperty {
+ /// Delete the enable property.
+ Enable,
+ /// Delete the mtu property.
+ Mtu,
+ /// Delete the comment property.
+ Comment,
+}
+
+#[api(
+ protected: true,
+ input: {
+ properties: {
+ name: {
+ schema: METRIC_SERVER_ID_SCHEMA,
+ },
+ update: {
+ type: InfluxDbUdpUpdater,
+ flatten: true,
+ },
+ delete: {
+ description: "List of properties to delete.",
+ type: Array,
+ optional: true,
+ items: {
+ type: DeletableProperty,
+ }
+ },
+ digest: {
+ optional: true,
+ schema: PROXMOX_CONFIG_DIGEST_SCHEMA,
+ },
+ },
+ },
+ returns: { type: InfluxDbUdp },
+ access: {
+ permission: &Permission::Privilege(&[], PRIV_SYS_MODIFY, false),
+ },
+)]
+/// Update an InfluxDB udp server configuration
+pub fn update_influxdb_udp_server(
+ name: String,
+ update: InfluxDbUdpUpdater,
+ delete: Option<Vec<DeletableProperty>>,
+ digest: Option<String>,
+ _rpcenv: &mut dyn RpcEnvironment,
+) -> Result<(), Error> {
+
+ let _lock = metrics::lock_config()?;
+
+ let (mut metrics, expected_digest) = metrics::config()?;
+
+ if let Some(ref digest) = digest {
+ let digest = <[u8; 32]>::from_hex(digest)?;
+ crate::tools::detect_modified_configuration_file(&digest, &expected_digest)?;
+ }
+
+ let mut config: InfluxDbUdp = metrics.lookup("influxdb-udp", &name)?;
+
+ if let Some(delete) = delete {
+ for delete_prop in delete {
+ match delete_prop {
+ DeletableProperty::Enable => { config.enable = None; },
+ DeletableProperty::Mtu => { config.mtu = None; },
+ DeletableProperty::Comment => { config.comment = None; },
+ }
+ }
+ }
+
+ if let Some(comment) = update.comment {
+ let comment = comment.trim().to_string();
+ if comment.is_empty() {
+ config.comment = None;
+ } else {
+ config.comment = Some(comment);
+ }
+ }
+
+ if let Some(host) = update.host { config.host = host; }
+ if let Some(port) = update.port { config.port = port; }
+
+ if update.enable.is_some() { config.enable = update.enable; }
+ if update.mtu.is_some() { config.mtu = update.mtu; }
+
+ metrics.set_data(&name, "influxdb-udp", &config)?;
+
+ metrics::save_config(&metrics)?;
+
+ Ok(())
+}
+
+const ITEM_ROUTER: Router = Router::new()
+ .get(&API_METHOD_READ_INFLUXDB_UDP_SERVER)
+ .put(&API_METHOD_UPDATE_INFLUXDB_UDP_SERVER)
+ .delete(&API_METHOD_DELETE_INFLUXDB_UDP_SERVER);
+
+pub const ROUTER: Router = Router::new()
+ .get(&API_METHOD_LIST_INFLUXDB_UDP_SERVERS)
+ .post(&API_METHOD_CREATE_INFLUXDB_UDP_SERVER)
+ .match_all("name", &ITEM_ROUTER);
diff --git a/src/api2/config/metricserver/mod.rs b/src/api2/config/metricserver/mod.rs
new file mode 100644
index 00000000..cbce34f7
--- /dev/null
+++ b/src/api2/config/metricserver/mod.rs
@@ -0,0 +1,16 @@
+use proxmox_router::{Router, SubdirMap};
+use proxmox_router::list_subdirs_api_method;
+use proxmox_sys::sortable;
+
+pub mod influxdbudp;
+pub mod influxdbhttp;
+
+#[sortable]
+const SUBDIRS: SubdirMap = &sorted!([
+ ("influxdb-http", &influxdbhttp::ROUTER),
+ ("influxdb-udp", &influxdbudp::ROUTER),
+]);
+
+pub const ROUTER: Router = Router::new()
+ .get(&list_subdirs_api_method!(SUBDIRS))
+ .subdirs(SUBDIRS);
diff --git a/src/api2/config/mod.rs b/src/api2/config/mod.rs
index c256ba64..5de1c28f 100644
--- a/src/api2/config/mod.rs
+++ b/src/api2/config/mod.rs
@@ -12,6 +12,7 @@ pub mod verify;
pub mod drive;
pub mod changer;
pub mod media_pool;
+pub mod metricserver;
pub mod tape_encryption_keys;
pub mod tape_backup_job;
pub mod traffic_control;
@@ -23,6 +24,7 @@ const SUBDIRS: SubdirMap = &[
("datastore", &datastore::ROUTER),
("drive", &drive::ROUTER),
("media-pool", &media_pool::ROUTER),
+ ("metricserver", &metricserver::ROUTER),
("remote", &remote::ROUTER),
("sync", &sync::ROUTER),
("tape-backup-job", &tape_backup_job::ROUTER),
--
2.30.2
^ permalink raw reply [flat|nested] 13+ messages in thread
* Re: [pbs-devel] [PATCH proxmox v3 2/3] proxmox-sys: add FileSystemInformation struct and helper
2021-12-17 8:09 ` [pbs-devel] [PATCH proxmox v3 2/3] proxmox-sys: add FileSystemInformation struct and helper Dominik Csapak
@ 2022-01-12 14:10 ` Wolfgang Bumiller
0 siblings, 0 replies; 13+ messages in thread
From: Wolfgang Bumiller @ 2022-01-12 14:10 UTC (permalink / raw)
To: Dominik Csapak; +Cc: pbs-devel
minor nit...
On Fri, Dec 17, 2021 at 09:09:53AM +0100, Dominik Csapak wrote:
> code mostly copied from proxmox-backups 'disk_usage'
>
> Signed-off-by: Dominik Csapak <d.csapak@proxmox.com>
> ---
> proxmox-sys/src/fs/mod.rs | 39 +++++++++++++++++++++++++++++++++++++++
> 1 file changed, 39 insertions(+)
>
> diff --git a/proxmox-sys/src/fs/mod.rs b/proxmox-sys/src/fs/mod.rs
> index 9fe333b..f8a7744 100644
> --- a/proxmox-sys/src/fs/mod.rs
> +++ b/proxmox-sys/src/fs/mod.rs
> @@ -102,3 +102,42 @@ impl CreateOptions {
> */
> }
>
> +/// Information about a mounted file system from statfs64 syscall
> +pub struct FileSystemInformation {
> + /// total bytes available
> + pub total: u64,
> + /// bytes used
> + pub used: u64,
> + /// bytes available to an unprivileged user
> + pub available: u64,
> + /// total number of inodes
> + pub total_inodes: u64,
> + /// free number of inodes
> + pub free_inodes: u64,
> + /// the type of the filesystem (see statfs64(2))
> + pub fs_type: i64,
> + /// the filesystem id
> + pub fs_id: libc::fsid_t,
> +}
> +
> +/// Get file system information from path
> +pub fn fs_info(path: &std::path::Path) -> Result<FileSystemInformation, Error> {
If you use `NixPath` below anyway, may as well go all the way:
pub fn fs_info<P: ?Sized + NixPath>(path: &P) -> nix::Result<FileSystemInformation> {
(note the error type change since that's the only kind of error
appearing in this function)
Otherwise, you could at least go the `std` way of `<P: AsRef<Path>>(path: P)`.
(We're already mixing these a bit, but `&Path` is the least convenient
way for callers to interact with)
> + let mut stat: libc::statfs64 = unsafe { std::mem::zeroed() };
> +
> + use nix::NixPath;
> +
> + let res = path.with_nix_path(|cstr| unsafe { libc::statfs64(cstr.as_ptr(), &mut stat) })?;
> + nix::errno::Errno::result(res)?;
> +
> + let bsize = stat.f_bsize as u64;
> +
> + Ok(FileSystemInformation {
> + total: stat.f_blocks * bsize,
> + used: (stat.f_blocks - stat.f_bfree) * bsize,
> + available: stat.f_bavail * bsize,
> + total_inodes: stat.f_files,
> + free_inodes: stat.f_ffree,
> + fs_type: stat.f_type,
> + fs_id: stat.f_fsid,
> + })
> +}
> --
> 2.30.2
^ permalink raw reply [flat|nested] 13+ messages in thread
* Re: [pbs-devel] [PATCH proxmox v3 3/3] proxmox-metrics: implement metrics server client code
2021-12-17 8:09 ` [pbs-devel] [PATCH proxmox v3 3/3] proxmox-metrics: implement metrics server client code Dominik Csapak
@ 2022-01-12 14:36 ` Wolfgang Bumiller
0 siblings, 0 replies; 13+ messages in thread
From: Wolfgang Bumiller @ 2022-01-12 14:36 UTC (permalink / raw)
To: Dominik Csapak; +Cc: pbs-devel
On Fri, Dec 17, 2021 at 09:09:54AM +0100, Dominik Csapak wrote:
> influxdb (udp + http(s)) only for now
>
> general architecture looks as follows:
>
> "new" returns a MetricsChannel and a Future
> the channels can be used to push data in (it flushes automatically if
> it would be over the configured size (mtu/max_body_size))
>
> and the future must be polled to actually send data to the servers.
>
> so most often it would look like this:
> let (future, channel) = InfluxDbHttp::new(..params..)?;
> let handle = tokio::spawn(future);
> channel.send_data(...).await?;
> handle.await?;
Sounds okay, but we already depend on tokio here, so maybe we should
consider always spawning and putting the spawn-handle and channel
together into a single helper.
The reason I'd like to do this is because I feel your above example is
not ideal. I'd prefer to enforce control over *aborting* the the spawned
task by wrapping it in `futures::future::abortable()` roughly like so:
struct Metrics {
join_handle: JoinHandle<_>,
abort_handle: AbortHandle,
channel: Sender<_>,
}
// convenience
impl Deref for Metrics {
type Target = Sender<_>;
fn deref(&self) -> &Sender { &self.channel }
}
impl Drop for Metrics {
fn drop(&mut self) {
self.abort_handle.abort();
}
}
impl Metrics {
pub async fn join(self) -> _ {
self.join_handle.join().await
}
}
>
> when all channels go out of scope, all remaining data in the channel
> will be read and sent to the server
>
> Signed-off-by: Dominik Csapak <d.csapak@proxmox.com>
> ---
> Cargo.toml | 1 +
> proxmox-metrics/Cargo.toml | 20 +++++
> proxmox-metrics/debian/changelog | 5 ++
> proxmox-metrics/debian/copyright | 16 ++++
> proxmox-metrics/debian/debcargo.toml | 7 ++
> proxmox-metrics/src/influxdb/http.rs | 122 ++++++++++++++++++++++++++
> proxmox-metrics/src/influxdb/mod.rs | 7 ++
> proxmox-metrics/src/influxdb/udp.rs | 94 ++++++++++++++++++++
> proxmox-metrics/src/influxdb/utils.rs | 50 +++++++++++
> proxmox-metrics/src/lib.rs | 89 +++++++++++++++++++
> 10 files changed, 411 insertions(+)
> create mode 100644 proxmox-metrics/Cargo.toml
> create mode 100644 proxmox-metrics/debian/changelog
> create mode 100644 proxmox-metrics/debian/copyright
> create mode 100644 proxmox-metrics/debian/debcargo.toml
> create mode 100644 proxmox-metrics/src/influxdb/http.rs
> create mode 100644 proxmox-metrics/src/influxdb/mod.rs
> create mode 100644 proxmox-metrics/src/influxdb/udp.rs
> create mode 100644 proxmox-metrics/src/influxdb/utils.rs
> create mode 100644 proxmox-metrics/src/lib.rs
>
> diff --git a/Cargo.toml b/Cargo.toml
> index 8f85e08..4a458d2 100644
> --- a/Cargo.toml
> +++ b/Cargo.toml
> @@ -6,6 +6,7 @@ members = [
> "proxmox-http",
> "proxmox-io",
> "proxmox-lang",
> + "proxmox-metrics",
> "proxmox-router",
> "proxmox-schema",
> "proxmox-serde",
> diff --git a/proxmox-metrics/Cargo.toml b/proxmox-metrics/Cargo.toml
> new file mode 100644
> index 0000000..9ac50fe
> --- /dev/null
> +++ b/proxmox-metrics/Cargo.toml
> @@ -0,0 +1,20 @@
> +[package]
> +name = "proxmox-metrics"
> +version = "0.1.0"
> +authors = ["Proxmox Support Team <support@proxmox.com>"]
> +edition = "2018"
> +license = "AGPL-3"
> +description = "Metrics Server export utilitites"
> +
> +exclude = [ "debian" ]
> +
> +[dependencies]
> +anyhow = "1.0"
> +tokio = { version = "1.0", features = [ "net", "sync" ] }
> +futures = "0.3"
> +serde = "1.0"
> +serde_json = "1.0"
> +http = "0.2"
> +hyper = "0.14"
> +openssl = "0.10"
> +proxmox-http = { path = "../proxmox-http", features = [ "client" ], version = "0.6" }
> diff --git a/proxmox-metrics/debian/changelog b/proxmox-metrics/debian/changelog
> new file mode 100644
> index 0000000..c02803b
> --- /dev/null
> +++ b/proxmox-metrics/debian/changelog
> @@ -0,0 +1,5 @@
> +rust-proxmox-metrics (0.1.0-1) unstable; urgency=medium
> +
> + * initial package
> +
> + -- Proxmox Support Team <support@proxmox.com> Tue, 14 Dec 2021 08:56:54 +0100
> diff --git a/proxmox-metrics/debian/copyright b/proxmox-metrics/debian/copyright
> new file mode 100644
> index 0000000..5661ef6
> --- /dev/null
> +++ b/proxmox-metrics/debian/copyright
> @@ -0,0 +1,16 @@
> +Copyright (C) 2021 Proxmox Server Solutions GmbH
> +
> +This software is written by Proxmox Server Solutions GmbH <support@proxmox.com>
> +
> +This program is free software: you can redistribute it and/or modify
> +it under the terms of the GNU Affero General Public License as published by
> +the Free Software Foundation, either version 3 of the License, or
> +(at your option) any later version.
> +
> +This program 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 Affero General Public License for more details.
> +
> +You should have received a copy of the GNU Affero General Public License
> +along with this program. If not, see <http://www.gnu.org/licenses/>.
> diff --git a/proxmox-metrics/debian/debcargo.toml b/proxmox-metrics/debian/debcargo.toml
> new file mode 100644
> index 0000000..b7864cd
> --- /dev/null
> +++ b/proxmox-metrics/debian/debcargo.toml
> @@ -0,0 +1,7 @@
> +overlay = "."
> +crate_src_path = ".."
> +maintainer = "Proxmox Support Team <support@proxmox.com>"
> +
> +[source]
> +vcs_git = "git://git.proxmox.com/git/proxmox.git"
> +vcs_browser = "https://git.proxmox.com/?p=proxmox.git"
> diff --git a/proxmox-metrics/src/influxdb/http.rs b/proxmox-metrics/src/influxdb/http.rs
> new file mode 100644
> index 0000000..df6c6a5
> --- /dev/null
> +++ b/proxmox-metrics/src/influxdb/http.rs
> @@ -0,0 +1,122 @@
> +use std::sync::Arc;
> +
> +use anyhow::{bail, Error};
> +use hyper::Body;
> +use openssl::ssl::{SslConnector, SslMethod, SslVerifyMode};
> +use tokio::sync::mpsc;
> +
> +use proxmox_http::client::{SimpleHttp, SimpleHttpOptions};
> +
> +use crate::influxdb::utils;
> +use crate::{MetricsChannel, MetricsData, MetricsServerFuture};
> +
> +pub struct InfluxDbHttp {
> + client: SimpleHttp,
> + _healthuri: http::Uri,
> + writeuri: http::Uri,
> + token: Option<String>,
> + max_body_size: usize,
> + data: String,
> + channel: mpsc::Receiver<Arc<MetricsData>>,
> +}
> +
> +impl InfluxDbHttp {
> + pub fn new(
> + https: bool,
> + host: &str,
> + port: u16,
> + organization: &str,
> + bucket: &str,
> + token: Option<&str>,
> + verify_tls: bool,
> + max_body_size: usize,
> + ) -> Result<(MetricsServerFuture, MetricsChannel), Error> {
> + let (tx, rx) = mpsc::channel(1024);
> +
> + let client = if verify_tls {
> + SimpleHttp::with_options(SimpleHttpOptions::default())
> + } else {
> + let mut ssl_connector = SslConnector::builder(SslMethod::tls()).unwrap();
> + ssl_connector.set_verify(SslVerifyMode::NONE);
> + SimpleHttp::with_ssl_connector(ssl_connector.build(), SimpleHttpOptions::default())
> + };
> +
> + let authority = proxmox_http::uri::build_authority(host, port)?;
> +
> + let writeuri = http::uri::Builder::new()
> + .scheme(if https { "https" } else { "http" })
> + .authority(authority.clone())
> + .path_and_query(format!(
> + "/api/v2/write?org={}&bucket={}",
> + organization, bucket
> + ))
> + .build()?;
> +
> + let healthuri = http::uri::Builder::new()
> + .scheme(if https { "https" } else { "http" })
> + .authority(authority)
> + .path_and_query("/health")
> + .build()?;
> +
> + let this = Self {
> + client,
> + writeuri,
> + _healthuri: healthuri,
> + token: token.map(String::from),
> + max_body_size,
> + data: String::new(),
> + channel: rx,
> + };
> +
> + let future = Box::pin(this.finish());
> + let channel = MetricsChannel { channel: tx };
> + Ok((future, channel))
> + }
> +
> + async fn add_data(&mut self, data: Arc<MetricsData>) -> Result<(), Error> {
> + let new_data = utils::format_influxdb_line(&data)?;
> +
> + if self.data.len() + new_data.len() >= self.max_body_size {
> + self.flush().await?;
> + }
> +
> + self.data.push_str(&new_data);
> +
> + if self.data.len() >= self.max_body_size {
> + self.flush().await?;
> + }
> +
> + Ok(())
> + }
> +
> + pub async fn flush(&mut self) -> Result<(), Error> {
> + if self.data.is_empty() {
> + return Ok(());
> + }
> + let mut request = http::Request::builder().method("POST").uri(&self.writeuri);
> +
> + if let Some(token) = &self.token {
> + request = request.header("Authorization", format!("Token {}", token));
> + }
> +
> + let request = request.body(Body::from(self.data.split_off(0)))?;
> +
> + let res = self.client.request(request).await?;
> +
> + let status = res.status();
> + if !status.is_success() {
> + bail!("got bad status: {}", status);
> + }
> + Ok(())
> + }
> +
> + async fn finish(mut self) -> Result<(), Error> {
> + while let Some(data) = self.channel.recv().await {
> + self.add_data(data).await?;
> + }
> +
> + self.flush().await?;
> +
> + Ok(())
> + }
> +}
> diff --git a/proxmox-metrics/src/influxdb/mod.rs b/proxmox-metrics/src/influxdb/mod.rs
> new file mode 100644
> index 0000000..26a715c
> --- /dev/null
> +++ b/proxmox-metrics/src/influxdb/mod.rs
> @@ -0,0 +1,7 @@
> +mod http;
> +pub use self::http::*;
> +
> +mod udp;
> +pub use udp::*;
> +
> +pub mod utils;
> diff --git a/proxmox-metrics/src/influxdb/udp.rs b/proxmox-metrics/src/influxdb/udp.rs
> new file mode 100644
> index 0000000..c4187a6
> --- /dev/null
> +++ b/proxmox-metrics/src/influxdb/udp.rs
> @@ -0,0 +1,94 @@
> +use std::sync::Arc;
> +
> +use anyhow::Error;
> +use tokio::net::UdpSocket;
> +use tokio::sync::mpsc;
> +
> +use crate::influxdb::utils;
> +use crate::{MetricsChannel, MetricsData, MetricsServerFuture};
> +
> +pub struct InfluxDbUdp {
> + address: String,
> + conn: Option<tokio::net::UdpSocket>,
> + ipv6: bool,
> + mtu: u16,
> + data: String,
> + channel: mpsc::Receiver<Arc<MetricsData>>,
> +}
> +
> +impl InfluxDbUdp {
> + pub fn new(host: &str, port: u16, mtu: Option<u16>) -> (MetricsServerFuture, MetricsChannel) {
^ Also this is a weird api, it would probably make more sense to just
have an `influx_db_udp()` function in this case, similar to how
`mpsc::channel()` is a function returning `(mpsc::Sender, mpsc::Receiver)`,
we'd have
metrics::influx_db_udp() -> Matrics;
metrics::influx_db_http() -> Matrics;
And yet another IPv6 thing...
Do we *really* want to specify host and port separately?
Why not have 1 string and use `ToSocketAddrs`?
> + let (tx, rx) = mpsc::channel(1024);
> + let ipv6 = host.contains(':');
The reason I'm asking is because the above works for ipv6 addresses...
but...
the schema you introduce in PBS 2/6 uses DNS_NAME_OR_IP_SCHEMA for the
host, so the hostname can be, you know, a *name*...
which means you don't actually *know* whether it's IPv4 or IPv6...
> +
> + let address = if ipv6 && host.len() > 3 && &host[0..1] != "[" {
> + format!("[{}]:{}", host, port)
> + } else {
> + format!("{}:{}", host, port)
> + };
> +
> + let this = Self {
> + address,
> + ipv6,
> + conn: None,
> + // empty ipv6 udp package needs 48 bytes, subtract 50 for safety
> + mtu: mtu.unwrap_or(1500) - 50,
> + data: String::new(),
> + channel: rx,
> + };
> +
> + let future = Box::pin(this.finish());
> +
> + let channel = MetricsChannel { channel: tx };
> +
> + (future, channel)
> + }
> +
> + async fn connect(&mut self) -> Result<tokio::net::UdpSocket, Error> {
Meaning this should actually iterate through the results of
tokio::net::lookup_host()
and do both bind() and connect() for each result one after the other
until they succeed...
> + let conn = if self.ipv6 {
> + UdpSocket::bind("[::]:0").await?
> + } else {
> + UdpSocket::bind("0.0.0.0:0").await?
> + };
> + let addr = self.address.clone();
> + conn.connect(addr).await?;
> + Ok(conn)
> + }
> +
> + async fn add_data(&mut self, data: Arc<MetricsData>) -> Result<(), Error> {
> + let new_data = utils::format_influxdb_line(&data)?;
> +
> + if self.data.len() + new_data.len() >= (self.mtu as usize) {
> + self.flush().await?;
> + }
> +
> + self.data.push_str(&new_data);
> +
> + if self.data.len() >= (self.mtu as usize) {
> + self.flush().await?;
> + }
> +
> + Ok(())
> + }
> +
> + async fn flush(&mut self) -> Result<(), Error> {
> + let conn = match self.conn.take() {
> + Some(conn) => conn,
> + None => self.connect().await?,
> + };
> +
> + conn.send(self.data.split_off(0).as_bytes()).await?;
> + self.conn = Some(conn);
> + Ok(())
> + }
> +
> + async fn finish(mut self) -> Result<(), Error> {
> + while let Some(data) = self.channel.recv().await {
> + self.add_data(data).await?;
> + }
> +
> + self.flush().await?;
> +
> + Ok(())
> + }
> +}
> diff --git a/proxmox-metrics/src/influxdb/utils.rs b/proxmox-metrics/src/influxdb/utils.rs
> new file mode 100644
> index 0000000..3507e61
> --- /dev/null
> +++ b/proxmox-metrics/src/influxdb/utils.rs
> @@ -0,0 +1,50 @@
> +use std::fmt::Write;
> +
> +use anyhow::{bail, Error};
> +use serde_json::Value;
> +
> +use crate::MetricsData;
> +
> +pub(crate) fn format_influxdb_line(data: &MetricsData) -> Result<String, Error> {
> + if !data.values.is_object() {
> + bail!("invalid data");
> + }
> +
> + let mut line = escape_measurement(&data.measurement);
> +
> + for (key, value) in &data.tags {
> + write!(line, ",{}={}", escape_key(key), escape_key(value))?;
> + }
> +
> + line.push(' ');
> +
> + let mut first = true;
> + for (key, value) in data.values.as_object().unwrap().iter() {
> + match value {
> + Value::Object(_) => bail!("objects not supported"),
> + Value::Array(_) => bail!("arrays not supported"),
> + _ => {}
> + }
> + if !first {
> + line.push(',');
> + }
> + first = false;
> + write!(line, "{}={}", escape_key(key), value.to_string())?;
> + }
> +
> + // nanosecond precision
> + writeln!(line, " {}", data.ctime * 1_000_000_000)?;
> +
> + Ok(line)
> +}
> +
> +fn escape_key(key: &str) -> String {
> + let key = key.replace(',', "\\,");
> + let key = key.replace('=', "\\=");
> + key.replace(' ', "\\ ")
> +}
> +
> +fn escape_measurement(measurement: &str) -> String {
> + let measurement = measurement.replace(',', "\\,");
> + measurement.replace(' ', "\\ ")
> +}
> diff --git a/proxmox-metrics/src/lib.rs b/proxmox-metrics/src/lib.rs
> new file mode 100644
> index 0000000..ba018fb
> --- /dev/null
> +++ b/proxmox-metrics/src/lib.rs
> @@ -0,0 +1,89 @@
> +use std::collections::HashMap;
> +use std::pin::Pin;
> +use std::sync::Arc;
> +
> +use anyhow::{format_err, Error};
> +use serde::Serialize;
> +use serde_json::Value;
> +use tokio::sync::mpsc;
> +
> +pub mod influxdb;
> +
> +#[derive(Clone)]
> +/// Structured data for the metric server
> +pub struct MetricsData {
> + /// The category of measurements
> + pub measurement: String,
> + /// A list of to attach to the measurements
> + pub tags: HashMap<String, String>,
> + /// The actual values to send. Only plain (not-nested) objects are supported at the moment.
> + pub values: Value,
> + /// The time of the measurement
> + pub ctime: i64,
> +}
> +
> +impl MetricsData {
> + /// Convenient helper to create from references
> + pub fn new<V: Serialize>(
> + measurement: &str,
> + tags: &[(&str, &str)],
> + ctime: i64,
> + values: V,
> + ) -> Result<Self, Error> {
> + let mut new_tags = HashMap::new();
> + for (key, value) in tags {
> + new_tags.insert(key.to_string(), value.to_string());
> + }
> +
> + Ok(Self {
> + measurement: measurement.to_string(),
> + tags: new_tags,
> + values: serde_json::to_value(values)?,
> + ctime,
> + })
> + }
> +}
> +
> +pub type MetricsServerFuture =
> + Pin<Box<dyn std::future::Future<Output = Result<(), Error>> + Send + 'static>>;
> +
> +#[derive(Clone)]
> +/// A channel to send data to the metric server
> +pub struct MetricsChannel {
> + pub(crate) channel: mpsc::Sender<Arc<MetricsData>>,
> +}
> +
> +impl MetricsChannel {
> + /// Queues the given data to the metric server
> + pub async fn send_data(&self, data: Arc<MetricsData>) -> Result<(), Error> {
> + // return ok if we got no data to send
> + if let Value::Object(map) = &data.values {
> + if map.is_empty() {
> + return Ok(());
> + }
> + }
> + self.channel
> + .send(data)
> + .await
> + .map_err(|_| format_err!("receiver side closed"))?;
> + Ok(())
> + }
> +}
> +
> +/// Helper to send a list of MetricsData to a list of MetricsChannels
> +pub async fn send_data_to_channels(
> + values: &[Arc<MetricsData>],
> + channels: &[MetricsChannel],
> +) -> Vec<Result<(), Error>> {
> + let mut futures = Vec::with_capacity(channels.len());
> + for channel in channels {
> + futures.push(async move {
> + for data in values {
> + channel.send_data(Arc::clone(data)).await?
> + }
> + Ok::<(), Error>(())
> + });
> + }
> +
> + futures::future::join_all(futures).await
> +}
> --
> 2.30.2
^ permalink raw reply [flat|nested] 13+ messages in thread
* Re: [pbs-devel] [PATCH proxmox-backup v3 2/6] pbs-api-types: add metrics api types
2021-12-17 8:09 ` [pbs-devel] [PATCH proxmox-backup v3 2/6] pbs-api-types: add metrics api types Dominik Csapak
@ 2022-01-12 14:40 ` Wolfgang Bumiller
0 siblings, 0 replies; 13+ messages in thread
From: Wolfgang Bumiller @ 2022-01-12 14:40 UTC (permalink / raw)
To: Dominik Csapak; +Cc: pbs-devel
On Fri, Dec 17, 2021 at 09:09:56AM +0100, Dominik Csapak wrote:
> InfluxDbUdp and InfluxDbHttp for now
>
> Signed-off-by: Dominik Csapak <d.csapak@proxmox.com>
> ---
> pbs-api-types/src/lib.rs | 2 +
> pbs-api-types/src/metrics.rs | 147 +++++++++++++++++++++++++++++++++++
> 2 files changed, 149 insertions(+)
> create mode 100644 pbs-api-types/src/metrics.rs
>
> diff --git a/pbs-api-types/src/lib.rs b/pbs-api-types/src/lib.rs
> index 0a0dd33d..09bd59c8 100644
> --- a/pbs-api-types/src/lib.rs
> +++ b/pbs-api-types/src/lib.rs
> @@ -88,6 +88,8 @@ pub use traffic_control::*;
> mod zfs;
> pub use zfs::*;
>
> +mod metrics;
> +pub use metrics::*;
>
> #[rustfmt::skip]
> #[macro_use]
> diff --git a/pbs-api-types/src/metrics.rs b/pbs-api-types/src/metrics.rs
> new file mode 100644
> index 00000000..786abe82
> --- /dev/null
> +++ b/pbs-api-types/src/metrics.rs
> @@ -0,0 +1,147 @@
> +use serde::{Deserialize, Serialize};
> +
> +use crate::{DNS_NAME_OR_IP_SCHEMA, PROXMOX_SAFE_ID_FORMAT, SINGLE_LINE_COMMENT_SCHEMA};
> +use proxmox_schema::{api, Schema, StringSchema, Updater};
> +
> +pub const METRIC_SERVER_ID_SCHEMA: Schema = StringSchema::new("Metrics Server ID.")
> + .format(&PROXMOX_SAFE_ID_FORMAT)
> + .min_length(3)
> + .max_length(32)
> + .schema();
> +
> +pub const INFLUXDB_BUCKET_SCHEMA: Schema = StringSchema::new("InfluxDB Bucket.")
> + .format(&PROXMOX_SAFE_ID_FORMAT)
> + .min_length(3)
> + .max_length(32)
> + .default("proxmox")
> + .schema();
> +
> +pub const INFLUXDB_ORGANIZATION_SCHEMA: Schema = StringSchema::new("InfluxDB Organization.")
> + .format(&PROXMOX_SAFE_ID_FORMAT)
> + .min_length(3)
> + .max_length(32)
> + .default("proxmox")
> + .schema();
> +
> +#[api(
> + properties: {
> + name: {
> + schema: METRIC_SERVER_ID_SCHEMA,
> + },
> + enable: {
> + description: "Enables or disables the metrics server",
^ same
I'd prefer not to mix doc-comments and `description` entries, please
stick to doc comments as they'll also show up in rustdocs.
> + type: bool,
> + optional: true,
> + default: true,
> + },
> + host: {
> + schema: DNS_NAME_OR_IP_SCHEMA,
> + },
> + mtu: {
> + description: "The MTU",
^ same
> + type: u16,
> + optional: true,
> + default: 1500,
> + },
> + comment: {
> + optional: true,
> + schema: SINGLE_LINE_COMMENT_SCHEMA,
> + },
> + },
> +)]
> +#[derive(Serialize, Deserialize, Updater)]
> +#[serde(rename_all = "kebab-case")]
> +/// InfluxDB Server (UDP)
> +pub struct InfluxDbUdp {
> + #[updater(skip)]
> + pub name: String,
and please put some newlines in between these, it's a bit much otherwise
with the #[serde] & doc comment stuff.
> + #[serde(skip_serializing_if = "Option::is_none")]
> + pub enable: Option<bool>,
> + pub host: String,
> + /// The port
> + pub port: u16,
> + #[serde(skip_serializing_if = "Option::is_none")]
> + pub mtu: Option<u16>,
> + #[serde(skip_serializing_if = "Option::is_none")]
> + pub comment: Option<String>,
> +}
> +
> +#[api(
> + properties: {
> + name: {
> + schema: METRIC_SERVER_ID_SCHEMA,
> + },
> + enable: {
> + description: "Enables or disables the metrics server",
^ same
> + type: bool,
> + optional: true,
> + default: true,
> + },
> + host: {
> + schema: DNS_NAME_OR_IP_SCHEMA,
> + },
> + https: {
> + description: "If true, HTTPS is used.",
^ same
> + type: bool,
> + optional: true,
> + default: true,
> + },
> + token: {
> + description: "The (optional) API token",
^ same, here for instance you used both...
> + type: String,
> + optional: true,
> + },
> + bucket: {
> + schema: INFLUXDB_BUCKET_SCHEMA,
> + optional: true,
> + },
> + organization: {
> + schema: INFLUXDB_ORGANIZATION_SCHEMA,
> + optional: true,
> + },
> + "max-body-size": {
> + description: "The (optional) maximum body size",
^ same
> + type: usize,
> + optional: true,
> + default: 25_000_000,
> + },
> + "verify-tls": {
> + description: "If true, the certificate will be validated.",
^ same
> + type: bool,
> + optional: true,
> + default: true,
> + },
> + comment: {
> + optional: true,
> + schema: SINGLE_LINE_COMMENT_SCHEMA,
> + },
> + },
> +)]
> +#[derive(Serialize, Deserialize, Updater)]
> +#[serde(rename_all = "kebab-case")]
> +/// InfluxDB Server (HTTP(s))
> +pub struct InfluxDbHttp {
> + #[updater(skip)]
> + pub name: String,
> + #[serde(skip_serializing_if = "Option::is_none")]
> + pub enable: Option<bool>,
> + pub host: String,
> + #[serde(skip_serializing_if = "Option::is_none")]
> + /// The (optional) port. (defaults: 80 for HTTP, 443 for HTTPS)
> + pub port: Option<u16>,
> + #[serde(skip_serializing_if = "Option::is_none")]
> + pub https: Option<bool>,
> + /// The Optional Token
> + #[serde(skip_serializing_if = "Option::is_none")]
> + pub token: Option<String>,
> + #[serde(skip_serializing_if = "Option::is_none")]
> + pub bucket: Option<String>,
> + #[serde(skip_serializing_if = "Option::is_none")]
> + pub organization: Option<String>,
> + #[serde(skip_serializing_if = "Option::is_none")]
> + pub max_body_size: Option<usize>,
> + #[serde(skip_serializing_if = "Option::is_none")]
> + pub verify_tls: Option<bool>,
> + #[serde(skip_serializing_if = "Option::is_none")]
> + pub comment: Option<String>,
> +}
> --
> 2.30.2
^ permalink raw reply [flat|nested] 13+ messages in thread
end of thread, other threads:[~2022-01-12 14:40 UTC | newest]
Thread overview: 13+ messages (download: mbox.gz / follow: Atom feed)
-- links below jump to the message on this page --
2021-12-17 8:09 [pbs-devel] [PATCH proxmox/proxmox-backup v3] add metrics server capability Dominik Csapak
2021-12-17 8:09 ` [pbs-devel] [PATCH proxmox v3 1/3] proxmox-sys: make some structs serializable Dominik Csapak
2021-12-17 8:09 ` [pbs-devel] [PATCH proxmox v3 2/3] proxmox-sys: add FileSystemInformation struct and helper Dominik Csapak
2022-01-12 14:10 ` Wolfgang Bumiller
2021-12-17 8:09 ` [pbs-devel] [PATCH proxmox v3 3/3] proxmox-metrics: implement metrics server client code Dominik Csapak
2022-01-12 14:36 ` Wolfgang Bumiller
2021-12-17 8:09 ` [pbs-devel] [PATCH proxmox-backup v3 1/6] use 'fs_info' from proxmox-sys Dominik Csapak
2021-12-17 8:09 ` [pbs-devel] [PATCH proxmox-backup v3 2/6] pbs-api-types: add metrics api types Dominik Csapak
2022-01-12 14:40 ` Wolfgang Bumiller
2021-12-17 8:09 ` [pbs-devel] [PATCH proxmox-backup v3 3/6] pbs-config: add metrics config class Dominik Csapak
2021-12-17 8:09 ` [pbs-devel] [PATCH proxmox-backup v3 4/6] backup-proxy: decouple stats gathering from rrd update Dominik Csapak
2021-12-17 8:09 ` [pbs-devel] [PATCH proxmox-backup v3 5/6] proxmox-backup-proxy: send metrics to configured metrics server Dominik Csapak
2021-12-17 8:10 ` [pbs-devel] [PATCH proxmox-backup v3 6/6] api: add metricserver endpoints Dominik Csapak
This is a public inbox, see mirroring instructions
for how to clone and mirror all data and code used for this inbox