proxmox-rrd/src/rrd.rs

   1 //! # Round Robin Database file format
   2
   3 use std::io::Read;
   4 use std::path::Path;
   5
   6 use anyhow::Error;
   7 use bitflags::bitflags;
   8
   9 use proxmox::tools::{fs::replace_file, fs::CreateOptions};
  10
  11 use proxmox_rrd_api_types::{RRDMode, RRDTimeFrameResolution};
  12
  13 /// The number of data entries per RRA
  14 pub const RRD_DATA_ENTRIES: usize = 70;
  15
  16 /// Proxmox RRD file magic number
  17 // openssl::sha::sha256(b"Proxmox Round Robin Database file v1.0")[0..8];
  18 pub const PROXMOX_RRD_MAGIC_1_0: [u8; 8] =  [206, 46, 26, 212, 172, 158, 5, 186];
  19
  20 use crate::DST;
  21
  22 bitflags!{
  23     /// Flags to specify the data soure type and consolidation function
  24     pub struct RRAFlags: u64 {
  25         // Data Source Types
  26         const DST_GAUGE  = 1;
  27         const DST_DERIVE = 2;
  28         const DST_COUNTER = 4;
  29         const DST_MASK   = 255; // first 8 bits
  30
  31         // Consolidation Functions
  32         const CF_AVERAGE = 1 << 8;
  33         const CF_MAX     = 2 << 8;
  34         const CF_MASK    = 255 << 8;
  35     }
  36 }
  37
  38 /// Round Robin Archive with [RRD_DATA_ENTRIES] data slots.
  39 ///
  40 /// This data structure is used inside [RRD] and directly written to the
  41 /// RRD files.
  42 #[repr(C)]
  43 pub struct RRA {
  44     /// Defined the data soure type and consolidation function
  45     pub flags: RRAFlags,
  46     /// Resulution (seconds) from [RRDTimeFrameResolution]
  47     pub resolution: u64,
  48     /// Last update time (epoch)
  49     pub last_update: f64,
  50     /// Count values computed inside this update interval
  51     pub last_count: u64,
  52     /// Stores the last value, used to compute differential value for derive/counters
  53     pub counter_value: f64,
  54     /// Data slots
  55     pub data: [f64; RRD_DATA_ENTRIES],
  56 }
  57
  58 impl RRA {
  59     fn new(flags: RRAFlags, resolution: u64) -> Self {
  60         Self {
  61             flags, resolution,
  62             last_update: 0.0,
  63             last_count: 0,
  64             counter_value: f64::NAN,
  65             data: [f64::NAN; RRD_DATA_ENTRIES],
  66         }
  67     }
  68
  69     fn delete_old(&mut self, time: f64) {
  70         let epoch = time as u64;
  71         let last_update = self.last_update as u64;
  72         let reso = self.resolution;
  73
  74         let min_time = epoch - (RRD_DATA_ENTRIES as u64)*reso;
  75         let min_time = (min_time/reso + 1)*reso;
  76         let mut t = last_update.saturating_sub((RRD_DATA_ENTRIES as u64)*reso);
  77         let mut index = ((t/reso) % (RRD_DATA_ENTRIES as u64)) as usize;
  78         for _ in 0..RRD_DATA_ENTRIES {
  79             t += reso; index = (index + 1) % RRD_DATA_ENTRIES;
  80             if t < min_time {
  81                 self.data[index] = f64::NAN;
  82             } else {
  83                 break;
  84             }
  85         }
  86     }
  87
  88     fn compute_new_value(&mut self, time: f64, value: f64) {
  89         let epoch = time as u64;
  90         let last_update = self.last_update as u64;
  91         let reso = self.resolution;
  92
  93         let index = ((epoch/reso) % (RRD_DATA_ENTRIES as u64)) as usize;
  94         let last_index = ((last_update/reso) % (RRD_DATA_ENTRIES as u64)) as usize;
  95
  96         if (epoch - (last_update as u64)) > reso || index != last_index {
  97             self.last_count = 0;
  98         }
  99
 100         let last_value = self.data[index];
 101         if last_value.is_nan() {
 102             self.last_count = 0;
 103         }
 104
 105         let new_count = if self.last_count < u64::MAX {
 106             self.last_count + 1
 107         } else {
 108             u64::MAX // should never happen
 109         };
 110
 111         if self.last_count == 0 {
 112             self.data[index] = value;
 113             self.last_count = 1;
 114         } else {
 115             let new_value = if self.flags.contains(RRAFlags::CF_MAX) {
 116                 if last_value > value { last_value } else { value }
 117             } else if self.flags.contains(RRAFlags::CF_AVERAGE) {
 118                 (last_value*(self.last_count as f64))/(new_count as f64)
 119                     + value/(new_count as f64)
 120             } else {
 121                 log::error!("rrdb update failed - unknown CF");
 122                 return;
 123             };
 124             self.data[index] = new_value;
 125             self.last_count = new_count;
 126         }
 127         self.last_update = time;
 128     }
 129
 130     fn update(&mut self, time: f64, mut value: f64, log_time_in_past: &mut bool) {
 131
 132         if time <= self.last_update {
 133             if *log_time_in_past {
 134                 log::warn!("rrdb update failed - time in past ({} < {})", time, self.last_update);
 135                 *log_time_in_past = false; // avoid logging this multiple times inside a RRD
 136             }
 137             return;
 138         }
 139
 140         if value.is_nan() {
 141             log::warn!("rrdb update failed - new value is NAN");
 142             return;
 143         }
 144
 145         // derive counter value
 146         if self.flags.intersects(RRAFlags::DST_DERIVE | RRAFlags::DST_COUNTER) {
 147             let time_diff = time - self.last_update;
 148             let is_counter = self.flags.contains(RRAFlags::DST_COUNTER);
 149
 150             let diff = if self.counter_value.is_nan() {
 151                 0.0
 152             } else if is_counter && value < 0.0 {
 153                 log::warn!("rrdb update failed - got negative value for counter");
 154                 return;
 155             } else if is_counter && value < self.counter_value {
 156                 // Note: We do not try automatic overflow corrections
 157                 self.counter_value = value;
 158                 log::warn!("rrdb update failed - conter overflow/reset detected");
 159                 return;
 160             } else {
 161                 value - self.counter_value
 162             };
 163             self.counter_value = value;
 164             value = diff/time_diff;
 165         }
 166
 167         self.delete_old(time);
 168         self.compute_new_value(time, value);
 169     }
 170 }
 171
 172 /// Round Robin Database file format with fixed number of [RRA]s
 173 #[repr(C)]
 174 // Note: Avoid alignment problems by using 8byte types only
 175 pub struct RRD {
 176     /// The magic number to identify the file type
 177     pub magic: [u8; 8],
 178     /// Hourly data (average values)
 179     pub hour_avg: RRA,
 180     /// Hourly data (maximum values)
 181     pub hour_max: RRA,
 182     /// Dayly data (average values)
 183     pub day_avg: RRA,
 184     /// Dayly data (maximum values)
 185     pub day_max: RRA,
 186     /// Weekly data (average values)
 187     pub week_avg: RRA,
 188     /// Weekly data (maximum values)
 189     pub week_max: RRA,
 190     /// Monthly data (average values)
 191     pub month_avg: RRA,
 192     /// Monthly data (maximum values)
 193     pub month_max: RRA,
 194     /// Yearly data (average values)
 195     pub year_avg: RRA,
 196     /// Yearly data (maximum values)
 197     pub year_max: RRA,
 198 }
 199
 200 impl RRD {
 201
 202     /// Create a new empty instance
 203     pub fn new(dst: DST) -> Self {
 204         let flags = match dst {
 205             DST::Gauge => RRAFlags::DST_GAUGE,
 206             DST::Derive => RRAFlags::DST_DERIVE,
 207         };
 208
 209         Self {
 210             magic: PROXMOX_RRD_MAGIC_1_0,
 211             hour_avg: RRA::new(
 212                 flags | RRAFlags::CF_AVERAGE,
 213                 RRDTimeFrameResolution::Hour as u64,
 214             ),
 215             hour_max: RRA::new(
 216                 flags |  RRAFlags::CF_MAX,
 217                 RRDTimeFrameResolution::Hour as u64,
 218             ),
 219             day_avg: RRA::new(
 220                 flags |  RRAFlags::CF_AVERAGE,
 221                 RRDTimeFrameResolution::Day as u64,
 222             ),
 223             day_max: RRA::new(
 224                 flags |  RRAFlags::CF_MAX,
 225                 RRDTimeFrameResolution::Day as u64,
 226             ),
 227             week_avg: RRA::new(
 228                 flags |  RRAFlags::CF_AVERAGE,
 229                 RRDTimeFrameResolution::Week as u64,
 230             ),
 231             week_max: RRA::new(
 232                 flags |  RRAFlags::CF_MAX,
 233                 RRDTimeFrameResolution::Week as u64,
 234             ),
 235             month_avg: RRA::new(
 236                 flags |  RRAFlags::CF_AVERAGE,
 237                 RRDTimeFrameResolution::Month as u64,
 238             ),
 239             month_max: RRA::new(
 240                 flags |  RRAFlags::CF_MAX,
 241                 RRDTimeFrameResolution::Month as u64,
 242             ),
 243             year_avg: RRA::new(
 244                 flags |  RRAFlags::CF_AVERAGE,
 245                 RRDTimeFrameResolution::Year as u64,
 246             ),
 247             year_max: RRA::new(
 248                 flags |  RRAFlags::CF_MAX,
 249                 RRDTimeFrameResolution::Year as u64,
 250             ),
 251         }
 252     }
 253
 254     /// Extract data from the archive
 255     pub fn extract_data(
 256         &self,
 257         time: f64,
 258         timeframe: RRDTimeFrameResolution,
 259         mode: RRDMode,
 260     ) -> (u64, u64, Vec<Option<f64>>) {
 261
 262         let epoch = time as u64;
 263         let reso = timeframe as u64;
 264
 265         let end = reso*(epoch/reso + 1);
 266         let start = end - reso*(RRD_DATA_ENTRIES as u64);
 267
 268         let mut list = Vec::new();
 269
 270         let raa = match (mode, timeframe) {
 271             (RRDMode::Average, RRDTimeFrameResolution::Hour) => &self.hour_avg,
 272             (RRDMode::Max, RRDTimeFrameResolution::Hour) => &self.hour_max,
 273             (RRDMode::Average, RRDTimeFrameResolution::Day) => &self.day_avg,
 274             (RRDMode::Max, RRDTimeFrameResolution::Day) => &self.day_max,
 275             (RRDMode::Average, RRDTimeFrameResolution::Week) => &self.week_avg,
 276             (RRDMode::Max, RRDTimeFrameResolution::Week) => &self.week_max,
 277             (RRDMode::Average, RRDTimeFrameResolution::Month) => &self.month_avg,
 278             (RRDMode::Max, RRDTimeFrameResolution::Month) => &self.month_max,
 279             (RRDMode::Average, RRDTimeFrameResolution::Year) => &self.year_avg,
 280             (RRDMode::Max, RRDTimeFrameResolution::Year) => &self.year_max,
 281         };
 282
 283         let rrd_end = reso*((raa.last_update as u64)/reso);
 284         let rrd_start = rrd_end - reso*(RRD_DATA_ENTRIES as u64);
 285
 286         let mut t = start;
 287         let mut index = ((t/reso) % (RRD_DATA_ENTRIES as u64)) as usize;
 288         for _ in 0..RRD_DATA_ENTRIES {
 289             if t < rrd_start || t > rrd_end {
 290                 list.push(None);
 291             } else {
 292                 let value = raa.data[index];
 293                 if value.is_nan() {
 294                     list.push(None);
 295                 } else {
 296                     list.push(Some(value));
 297                 }
 298             }
 299             t += reso; index = (index + 1) % RRD_DATA_ENTRIES;
 300         }
 301
 302         (start, reso, list)
 303     }
 304
 305     /// Create instance from raw data, testing data len and magic number
 306     pub fn from_raw(mut raw: &[u8]) -> Result<Self, std::io::Error> {
 307         let expected_len = std::mem::size_of::<RRD>();
 308         if raw.len() != expected_len {
 309             let msg = format!("wrong data size ({} != {})", raw.len(), expected_len);
 310             return Err(std::io::Error::new(std::io::ErrorKind::Other, msg));
 311         }
 312
 313         let mut rrd: RRD = unsafe { std::mem::zeroed() };
 314         unsafe {
 315             let rrd_slice = std::slice::from_raw_parts_mut(&mut rrd as *mut _ as *mut u8, expected_len);
 316             raw.read_exact(rrd_slice)?;
 317         }
 318
 319         if rrd.magic != PROXMOX_RRD_MAGIC_1_0 {
 320             let msg = "wrong magic number".to_string();
 321             return Err(std::io::Error::new(std::io::ErrorKind::Other, msg));
 322         }
 323
 324         Ok(rrd)
 325     }
 326
 327     /// Load data from a file
 328     pub fn load(path: &Path) -> Result<Self, std::io::Error> {
 329         let raw = std::fs::read(path)?;
 330         Self::from_raw(&raw)
 331     }
 332
 333     /// Store data into a file (atomic replace file)
 334     pub fn save(&self, filename: &Path, options: CreateOptions) -> Result<(), Error> {
 335         let rrd_slice = unsafe {
 336             std::slice::from_raw_parts(self as *const _ as *const u8, std::mem::size_of::<RRD>())
 337         };
 338         replace_file(filename, rrd_slice, options)
 339     }
 340
 341     /// Update the value (in memory)
 342     ///
 343     /// Note: This does not call [Self::save].
 344     pub fn update(&mut self, time: f64, value: f64) {
 345
 346         if value.is_nan() {
 347             log::warn!("rrdb update failed - new value is NAN");
 348             return;
 349         }
 350
 351         let mut log_time_in_past = true;
 352
 353         self.hour_avg.update(time, value, &mut log_time_in_past);
 354         self.hour_max.update(time, value, &mut log_time_in_past);
 355
 356         self.day_avg.update(time, value, &mut log_time_in_past);
 357         self.day_max.update(time, value, &mut log_time_in_past);
 358
 359         self.week_avg.update(time, value, &mut log_time_in_past);
 360         self.week_max.update(time, value, &mut log_time_in_past);
 361
 362         self.month_avg.update(time, value, &mut log_time_in_past);
 363         self.month_max.update(time, value, &mut log_time_in_past);
 364
 365         self.year_avg.update(time, value, &mut log_time_in_past);
 366         self.year_max.update(time, value, &mut log_time_in_past);
 367     }
 368 }