Always prune old values for the aggregator, even during insertion
@@ -437,19 +437,6 @@ impl AggregatedCounters {
}
}
-pub async fn aggregate(dec: &Decision, rcode: Option<u32>, rinfo: &RequestInfo, tags: &Tags) {
- let seconds = rinfo.timestamp.timestamp();
- let key = AggregationKey {
- proxy: None,
- secpolid: rinfo.rinfo.policyid.to_string(),
- secpolentryid: rinfo.rinfo.entryid.to_string(),
- };
- let mut guard = AGGREGATED.lock().await;
- let entry_secs = guard.entry(seconds).or_default();
- let entry = entry_secs.entry(key).or_default();
- entry.increment(dec, rcode, rinfo, tags);
-}
-
/* missing:
* d-bytes
@@ -584,22 +571,23 @@ fn serialize_entry(secs: i64, hdr: &AggregationKey, counters: &AggregatedCounter
Value::Object(content)
}
-/// displays the Nth last seconds of aggregated data
-pub async fn aggregated_values() -> String {
- let mut guard = AGGREGATED.lock().await;
- // first, prune excess data
- let keys: Vec<i64> = guard.keys().copied().collect();
- match keys.last() {
- None => return "[]".into(),
- Some(last_entry) => {
+fn prune_old_values<A>(mp: &mut BTreeMap<i64, A>) {
+ let keys: Vec<i64> = mp.keys().copied().collect();
+ if let Some(last_entry) = keys.last() {
for k in keys.iter() {
if *k < *last_entry - *SECONDS_KEPT {
- guard.remove(k);
+ mp.remove(k);
}
}
}
}
+/// displays the Nth last seconds of aggregated data
+pub async fn aggregated_values() -> String {
+ let mut guard = AGGREGATED.lock().await;
+ // first, prune excess data
+ prune_old_values(&mut guard);
+
let entries: Vec<Value> = guard
.iter()
.flat_map(|(secs, v)| {
@@ -615,3 +603,18 @@ pub async fn aggregated_values() -> String {
pub fn aggregated_values_block() -> String {
async_std::task::block_on(aggregated_values())
}
+
+/// adds new data to the aggregator
+pub async fn aggregate(dec: &Decision, rcode: Option<u32>, rinfo: &RequestInfo, tags: &Tags) {
+ let seconds = rinfo.timestamp.timestamp();
+ let key = AggregationKey {
+ proxy: None,
+ secpolid: rinfo.rinfo.policyid.to_string(),
+ secpolentryid: rinfo.rinfo.entryid.to_string(),
+ };
+ let mut guard = AGGREGATED.lock().await;
+ prune_old_values(&mut guard);
+ let entry_secs = guard.entry(seconds).or_default();
+ let entry = entry_secs.entry(key).or_default();
+ entry.increment(dec, rcode, rinfo, tags);
+}
