Add metrics and logs

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "bioyino"
-version = "0.8.0"
+version = "0.8.2"
 authors = ["Sergey Noskov aka Albibek <[email protected]>"]
 description = "StatsD-compatible, high-performance, fault-tolerant metric aggregator"
 edition = "2018"

src/aggregate.rs

@@ -1,4 +1,4 @@
-use std::collections::HashMap;
+use std::{collections::HashMap, sync::atomic::Ordering};
 use std::sync::Arc;
 
 use async_channel::Sender as AsyncSender;
@@ -12,7 +12,7 @@ use bioyino_metric::{
     FromF64,
 };
 
-use crate::cache::RotatedCacheShard;
+use crate::{cache::RotatedCacheShard, stats::STATS};
 use crate::config::{all_aggregates, Aggregation, ConfigError, Naming, RoundTimestamp};
 use crate::{s, Float};
 
@@ -164,9 +164,10 @@ pub fn aggregate_task(data: AggregationData) {
                     _ => Some((name.clone(), typename, *aggregate, value)),
                 }
             })
-        .map(|data| result.push(data))
-            .last();
-        }
+            .map(|data| result.push(data))
+                .last();
+    }
+    STATS.aggregated_metrics.fetch_add(result.len(), Ordering::Relaxed);
 
     futures::executor::block_on(response.send(result)).unwrap_or(());
 }

src/carbon.rs

@@ -291,6 +291,7 @@ impl CarbonBackend {
         let mut writer = CarbonCodec::new(ts.clone(), options.agg.clone()).framed(conn);
 
         for m in metrics.iter().flatten() {
+            s!(egress_carbon_metrics);
             writer.send(m).await?
         }
 

src/main.rs

@@ -66,7 +66,7 @@ pub type Cache = HashMap<MetricName, Metric<Float>>;
 pub static CONSENSUS_STATE: Lazy<Mutex<ConsensusState>> = Lazy::new(|| Mutex::new(ConsensusState::Disabled));
 pub static IS_LEADER: AtomicBool = AtomicBool::new(false);
 
-fn main() {
+fn main() { 
     let (system, command) = System::load().expect("loading config");
 
     let config = system.clone();
@@ -170,7 +170,11 @@ fn main() {
     let stats_prefix = stats_prefix.trim_end_matches('.').to_string();
     // Spawn future gatering bioyino own stats
     info!(own_stat_log, "starting own stats counter");
-    let own_stats = OwnStats::new(s_interval, stats_prefix, slow_chan.clone(), fast_prio_chans.clone(), own_stat_log);
+    let own_stats = OwnStats::new(
+        s_interval, stats_prefix, slow_chan.clone(), 
+        fast_prio_chans.clone(), own_stat_log, carbon.clone(),
+        aggregation.clone(), naming.clone(),
+    );
     runtime.spawn(own_stats.run());
 
     let compat_log = rlog.new(o!("thread" => "compat"));

src/slow_task.rs

@@ -1,16 +1,17 @@
 use std::sync::Arc;
+use std::sync::atomic::Ordering;
 
 use crossbeam_channel::Sender;
 use futures::channel::oneshot;
-use log::warn as logw;
-use slog::{error, info, Logger};
+use slog::{error, info, Logger, trace};
 
 use bioyino_metric::{name::MetricName, Metric, MetricTypeName};
 
 use crate::aggregate::{aggregate_task, AggregationData};
 use crate::cache::{RotatedCache, SharedCache};
 use crate::config::System;
 
+use crate::stats::STATS;
 use crate::{s, Cache, Float};
 
 #[derive(Debug)]
@@ -68,7 +69,8 @@ impl SlowTaskRunner {
         let ename = name.clone();
         let em = MetricTypeName::from_metric(&metric);
         self.cache.accumulate(name, metric).unwrap_or_else(|_| {
-            logw!(
+            trace!(
+                self.log,
                 "could not accumulate in long cache at {:?} new type '{}'",
                 String::from_utf8_lossy(&ename.name[..]),
                 em.to_string(),
@@ -84,13 +86,20 @@ impl SlowTaskRunner {
                 list.drain(..).map(|(name, metric)| self.update_metric(name, metric)).last();
             }
             SlowTask::Join(cache) => {
-                cache.iter().map(|(name, metric)| self.update_metric(name.clone(), metric.clone())).last();
+                cache.iter().map(
+                    |(name, metric)| {
+                        s!(slow_cache_joined_metrics);
+                        self.update_metric(name.clone(), metric.clone());
+                    }
+                ).last();
             }
             SlowTask::AddSnapshot(mut list) => {
                 list.drain(..).map(|(name, metric)| self.update_metric(name, metric)).last();
             }
             SlowTask::Rotate(channel) => {
                 let rotated = self.cache.rotate(channel.is_some());
+                let sum: usize = rotated.iter().map(|m| m.len()).sum();
+                STATS.slow_cache_rotated_metrics.fetch_add(sum, Ordering::Relaxed);
                 if let Some(c) = channel {
                     let log = self.log.clone();
                     c.send(rotated).unwrap_or_else(|_| {