← 返回文本›编辑 / 分支

Hew metrics aggregation demo

#9GkN25dHc6public / public新版本

由 @slepp 已创建 5 days ago 永不过期 2.8 KB 语法: hew

你的更改会创建一个链接到此文本的新文本 — 原始文本不变。

新版本 你的更改会创建一个链接到此文本的新文本 — 原始文本不变。

可见性公开动态访问 public

变更备注 (可选)

此文本会显示在公开动态中。如果只想通过链接分享，请更改可见性。

取消

//! 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)}");
}