implement generic observable
This commit is contained in:
parent
822622d9a9
commit
cfa70c946b
|
|
@ -938,9 +938,6 @@ dependencies = [
|
||||||
[[package]]
|
[[package]]
|
||||||
name = "protocol"
|
name = "protocol"
|
||||||
version = "0.1.0"
|
version = "0.1.0"
|
||||||
dependencies = [
|
|
||||||
"regex",
|
|
||||||
]
|
|
||||||
|
|
||||||
[[package]]
|
[[package]]
|
||||||
name = "pxfm"
|
name = "pxfm"
|
||||||
|
|
|
||||||
|
|
@ -2,6 +2,3 @@
|
||||||
name = "protocol"
|
name = "protocol"
|
||||||
version = "0.1.0"
|
version = "0.1.0"
|
||||||
edition = "2024"
|
edition = "2024"
|
||||||
|
|
||||||
[dependencies]
|
|
||||||
regex = "1"
|
|
||||||
|
|
|
||||||
|
|
@ -1,8 +1,6 @@
|
||||||
use std::io::{Read, Write};
|
use std::io::{Read, Write};
|
||||||
use std::os::unix::net::UnixStream;
|
use std::os::unix::net::UnixStream;
|
||||||
|
|
||||||
use regex::Regex;
|
|
||||||
|
|
||||||
use crate::err::ProtocolError;
|
use crate::err::ProtocolError;
|
||||||
use crate::message::{
|
use crate::message::{
|
||||||
create_list_message, create_request_message, parse_list_response, parse_response_header,
|
create_list_message, create_request_message, parse_list_response, parse_response_header,
|
||||||
|
|
@ -48,15 +46,10 @@ impl Protocol {
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
pub fn list(&mut self, path_regex: &Regex) -> Result<Vec<String>, ProtocolError> {
|
pub fn list(&mut self) -> Result<Vec<String>, ProtocolError> {
|
||||||
let req = create_list_message();
|
let req = create_list_message();
|
||||||
let lines = self.rpc(&req)?;
|
let lines = self.rpc(&req)?;
|
||||||
let values = parse_list_response(lines);
|
Ok(parse_list_response(lines))
|
||||||
|
|
||||||
Ok(values
|
|
||||||
.into_iter()
|
|
||||||
.filter(|v| path_regex.is_match(v))
|
|
||||||
.collect())
|
|
||||||
}
|
}
|
||||||
|
|
||||||
fn rpc(&mut self, req: &str) -> Result<Vec<String>, ProtocolError> {
|
fn rpc(&mut self, req: &str) -> Result<Vec<String>, ProtocolError> {
|
||||||
|
|
|
||||||
|
|
@ -1,7 +1,7 @@
|
||||||
mod observables;
|
mod observables;
|
||||||
|
|
||||||
use draw::{StackedConfig, StackedSeries, StackedSeriesConfig, stacked};
|
use draw::{StackedConfig, StackedSeries, StackedSeriesConfig, stacked};
|
||||||
use observables::CpuObservable;
|
use observables::{Observable, ObservableConfig};
|
||||||
use protocol::Protocol;
|
use protocol::Protocol;
|
||||||
use std::{thread, time};
|
use std::{thread, time};
|
||||||
|
|
||||||
|
|
@ -19,7 +19,13 @@ fn main() {
|
||||||
|
|
||||||
if let Ok(mut proto) = Protocol::new("/var/run/collectd-unixsock") {
|
if let Ok(mut proto) = Protocol::new("/var/run/collectd-unixsock") {
|
||||||
// loop {
|
// loop {
|
||||||
let cpu = CpuObservable::discover(&mut proto).unwrap();
|
|
||||||
|
let cpu_load_cfg = ObservableConfig {
|
||||||
|
name: "cpu".to_string(),
|
||||||
|
norm: 1.,
|
||||||
|
include: vec![Regex::new()],
|
||||||
|
};
|
||||||
|
let cpu = Observable::discover(&mut proto).unwrap();
|
||||||
let mut series = StackedSeries::new(cfg);
|
let mut series = StackedSeries::new(cfg);
|
||||||
for _ in 1..10 {
|
for _ in 1..10 {
|
||||||
let sample = cpu.sample(&mut proto).unwrap();
|
let sample = cpu.sample(&mut proto).unwrap();
|
||||||
|
|
|
||||||
|
|
@ -1,55 +1,43 @@
|
||||||
use protocol::{Protocol, ProtocolError};
|
use protocol::{Protocol, ProtocolError};
|
||||||
use regex::Regex;
|
use regex::Regex;
|
||||||
|
|
||||||
const CPU_NORM: f32 = 100.;
|
pub struct ObservableConfig {
|
||||||
|
|
||||||
pub struct CpuObservable {
|
|
||||||
pub name: String,
|
pub name: String,
|
||||||
num_cpus: u8,
|
pub norm: f32,
|
||||||
wait_cpus: Vec<String>,
|
pub include: Vec<Regex>,
|
||||||
interrupt_cpus: Vec<String>,
|
|
||||||
load_cpus: Vec<String>,
|
|
||||||
idle_cpus: Vec<String>,
|
|
||||||
}
|
}
|
||||||
|
|
||||||
impl CpuObservable {
|
pub struct Observable {
|
||||||
pub fn discover(proto: &mut Protocol) -> Result<CpuObservable, ProtocolError> {
|
pub name: String,
|
||||||
let wait_cpus = proto.list(&Regex::new(".*\\/cpu-[0-9]+\\/cpu-wait").unwrap())?;
|
norm: f32,
|
||||||
let interrupt_cpus =
|
metrics: Vec<String>,
|
||||||
proto.list(&Regex::new(".*\\/cpu-[0-9]+\\/cpu-(softirq|interrupt)").unwrap())?;
|
}
|
||||||
let load_cpus = proto.list(&Regex::new(".*\\/cpu-[0-9]+\\/cpu-.+").unwrap())?;
|
|
||||||
let idle_cpus = proto.list(&Regex::new(".*\\/cpu-[0-9]+\\/cpu-idle").unwrap())?;
|
|
||||||
|
|
||||||
let num_cpus = wait_cpus.iter().len();
|
impl Observable {
|
||||||
|
pub fn discover(
|
||||||
|
proto: &mut Protocol,
|
||||||
|
cfg: ObservableConfig,
|
||||||
|
) -> Result<Observable, ProtocolError> {
|
||||||
|
let all_values = proto.list()?;
|
||||||
|
let filtered_values = all_values
|
||||||
|
.into_iter()
|
||||||
|
.filter(|v| cfg.include.iter().any(|i| i.is_match(v)))
|
||||||
|
.collect();
|
||||||
|
|
||||||
Ok(CpuObservable {
|
Ok(Observable {
|
||||||
name: "cpu".into(),
|
name: cfg.name,
|
||||||
num_cpus: num_cpus as u8,
|
norm: cfg.norm,
|
||||||
wait_cpus: wait_cpus,
|
metrics: filtered_values,
|
||||||
interrupt_cpus: interrupt_cpus,
|
|
||||||
load_cpus: load_cpus,
|
|
||||||
idle_cpus: idle_cpus,
|
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
|
|
||||||
pub fn sample(&self, proto: &mut Protocol) -> Result<Vec<f32>, ProtocolError> {
|
pub fn sample(&self, proto: &mut Protocol) -> Result<f32, ProtocolError> {
|
||||||
let load = self.sample_fields(&self.load_cpus, proto)?;
|
let results = self
|
||||||
let iowait = self.sample_fields(&self.wait_cpus, proto)?;
|
.metrics
|
||||||
let interrupts = self.sample_fields(&self.interrupt_cpus, proto)?;
|
|
||||||
let idle = self.sample_fields(&self.idle_cpus, proto)?;
|
|
||||||
|
|
||||||
Ok(vec![load - iowait - interrupts - idle, iowait, interrupts])
|
|
||||||
}
|
|
||||||
|
|
||||||
fn sample_fields(
|
|
||||||
&self,
|
|
||||||
fields: &Vec<String>,
|
|
||||||
proto: &mut Protocol,
|
|
||||||
) -> Result<f32, ProtocolError> {
|
|
||||||
let results = fields
|
|
||||||
.iter()
|
.iter()
|
||||||
.map(|f| proto.get(f.into(), "value".to_string()))
|
.map(|f| proto.get(f.into(), "value".to_string()))
|
||||||
.collect::<Result<Vec<f32>, ProtocolError>>()?;
|
.collect::<Result<Vec<f32>, ProtocolError>>()?;
|
||||||
Ok(results.into_iter().sum::<f32>() / self.num_cpus as f32 / CPU_NORM)
|
|
||||||
|
Ok(results.into_iter().sum::<f32>() / self.norm)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
|
||||||
Loading…
Reference in New Issue