Meter.pudu

Pudu62 lines2.6 KB

Copy pathGitHub ↗

1/** @Telemetry.Meter.Module — counts and durations gathered from events */2module PuduLangResilience.Telemetry.Meter34import Std.Map as Map5import PuduLangResilience.Telemetry as Telemetry6import PuduLangResilience.Utils.Shared as Shared78/** @Telemetry.Meter.Meter — event counts and durations by series */9export type Meter = {10  counts: Shared.Shared[Map[Str, Int]],11  durations: Shared.Shared[Map[Str, Array[Int]]],12  enrich: fn(Telemetry.Event) -> Array[(Str, Str)]13}1415/// A meter naming each series by event name and source.16export fn create() -> Meter { enriched(fn(_event: Telemetry.Event) -> Array[(Str, Str)] { [] }) }1718/// A meter that also names each series by the tags `enrich` answers for its event.19export fn enriched(enrich: fn(Telemetry.Event) -> Array[(Str, Str)]) -> Meter {20  Meter{counts: Shared.shared(Map.empty()), durations: Shared.shared(Map.empty()), enrich: enrich}21}2223/// A listener counting every event into the meter, and recording the duration of every attempt24/// and every finished execution.25export fn listener(meter: &Meter) -> Telemetry.Listener {26  let target = *meter27  fn(event: Telemetry.Event) -> () {28    let label = seriesOf(&target, &event)29    Shared.update(&target.counts, fn(held: Map[Str, Int]) -> Map[Str, Int] { Map.upsert(&held, label, 1, |seen: Int| seen + 1) })30    let measured = match event.detail {31      case Telemetry.Attempted(_, duration, _) => Some(duration)32      case Telemetry.Executed(duration) => Some(duration)33      case _ => None34    }35    if let Some(duration) = measured {36      Shared.update(&target.durations, fn(held: Map[Str, Array[Int]]) -> Map[Str, Array[Int]] { Map.upsert(&held, event.name, [duration], |seen: Array[Int]| seen.push(duration)) })37    }38  }39}4041/// How many events of the given name were counted, over every series.42export fn count(meter: &Meter, name: Str) -> Int {43  var total = 044  for entry in Map.pairs(&Shared.current(&meter.counts)) {45    if entry[0] == name || entry[0].startsWith(name + " ") { total = total + entry[1] }46  }47  total48}4950/// Every series and its count, in order.51export fn series(meter: &Meter) -> Array[(Str, Int)] { Map.pairs(&Shared.current(&meter.counts)) }5253/// Every duration recorded for events of the given name, in the order they were reported.54export fn durations(meter: &Meter, name: Str) -> Array[Int] { Map.getOr(&Shared.current(&meter.durations), name, []) }5556/// The series an event is counted in: its name, its source, and its enriched tags.57fn seriesOf(meter: &Meter, event: &Telemetry.Event) -> Str {58  var tags = ["source=" + Telemetry.sourceName(&event.source)]59  for tag in (meter.enrich)(*event) { tags = tags.push(tag[0] + "=" + tag[1]) }60  event.name + " " + tags.join(",")61}62