← back to paste›Edit / fork

Hew metrics aggregation demo

#9GkN25dHc6public / publicnew version

by @slepp created 5 days ago no expiry 2.8 KB syntax: hew

Your changes create a new paste linked to this one — the original is untouched.

new version Your changes create a new paste linked to this one — the original is untouched.

Visibility Public feed Access public

Change note (optional)

This paste will be listed on the public feed. Change Visibility if you only want people with the link to see it.

Cancel

//! A small metrics aggregation service: sources push samples to a
//! collector actor, a periodic ticker counts snapshots, and asks read
//! the aggregates back out.

type Sample {
    name: string;
    value: f64;
}

actor Collector {
    var counts: HashMap<string, i64> = HashMap::new();
    var sums: HashMap<string, f64> = HashMap::new();
    var snapshots_taken: i64 = 0;

receive fn observe(name: string, value: f64) {
        let prior_count = match counts.get(name) {
            Some(c) => c,
            None => 0,
        };
        counts.insert(name, prior_count + 1);
        let prior_sum = match sums.get(name) {
            Some(s) => s,
            None => 0.0,
        };
        sums.insert(name, prior_sum + value);
    }

#[every(40ms)]
    receive fn snapshot() {
        snapshots_taken = snapshots_taken + 1;
    }

receive fn mean(name: string) -> f64 {
        let c = match counts.get(name) {
            Some(c) => c,
            None => 0,
        };
        if c == 0 {
            return 0.0;
        }
        let s = match sums.get(name) {
            Some(s) => s,
            None => 0.0,
        };
        s / (c as f64)
    }

receive fn count(name: string) -> i64 {
        match counts.get(name) {
            Some(c) => c,
            None => 0,
        }
    }

receive fn snapshots() -> i64 {
        snapshots_taken
    }
}

fn percentile_label(p: f64) -> string {
    match p {
        v if v >= 0.99 => "p99",
        v if v >= 0.95 => "p95",
        v if v >= 0.5 => "p50",
        _ => "low",
    }
}

fn main() {
    let collector = spawn Collector();

// Sends from inside a function closure are not yet implemented
    // (captured pid loses its actor layout in the closure's MIR context);
    // push samples from plain loops instead.
    var i = 0;
    while i < 100 {
        collector.observe("api.latency", 10.0 + ((i % 50) as f64));
        i = i + 1;
    }
    var j = 0;
    while j < 50 {
        collector.observe("db.latency", 3.0 + (j as f64));
        j = j + 1;
    }

sleep_ms(200);

match await collector.count("api.latency") {
        Ok(n) => println(f"api.latency samples: {n}"),
        Err(_) => println("ask failed: count"),
    }
    match await collector.mean("api.latency") {
        Ok(m) => println(f"api.latency mean: {m}"),
        Err(_) => println("ask failed: mean"),
    }
    match await collector.mean("db.latency") {
        Ok(m) => println(f"db.latency mean: {m}"),
        Err(_) => println("ask failed: mean"),
    }
    match await collector.snapshots() {
        Ok(s) => {
            if s >= 3 {
                println("periodic snapshots: ok");
            } else {
                println(f"periodic snapshots: only {s}");
            }
        },
        Err(_) => println("ask failed: snapshots"),
    }
    println(f"label: {percentile_label(0.97)}");
}