Updates

measurement-rate-control/hyper-p2p-histogram-gossip/index.js

@@ -1,52 +1,148 @@
 require('bare-process/global')
 const EventEmitter = require('bare-events')
+const b4a = require('b4a')
 const { assertNonEmpty } = require('../../_shared/lib/errors.js')
-const { attachGossip, sendGossip } = require('../../_shared/storage-gossip-base.js')
+const { initModuleSwarm, gossipSend } = require('../../_shared/p2p-bare.js')
 
 const PROTOCOL = 'histogram-gossip/v1'
+const DEFAULT_BUCKETS = 10
+
+function bucketIndex (value, min, max, count) {
+  if (value <= min) return 0
+  if (value >= max) return count - 1
+  const span = (max - min) / count
+  return Math.min(count - 1, Math.floor((value - min) / span))
+}
 
 class HyperP2PHistogramGossip extends EventEmitter {
   constructor (opts = {}) {
     super()
     this.topic = opts.topic || null
     this.keyPair = opts.keyPair || require('hypercore-crypto').keyPair()
-    this._samples = []
-    this._stats = { samples: 0, gossipIn: 0, gossipOut: 0 }
+    this.peerHex = b4a.toString(this.keyPair.publicKey, 'hex')
+    this._bucketCount = opts.buckets || DEFAULT_BUCKETS
+    this._metrics = new Map()
+    this._stats = { observed: 0, gossipIn: 0, gossipOut: 0 }
     this.swarm = null
+    this._peerMsgs = null
   }
 
-  record (value) {
-    if (typeof value !== 'number') throw new Error('value must be a number')
-    this._samples.push({ value, at: Date.now() })
-    this._stats.samples++
-    sendGossip(this, { type: 'histogram-gossip-sync', value })
+  _ensure (metric) {
+    if (!this._metrics.has(metric)) {
+      this._metrics.set(metric, {
+        min: Infinity,
+        max: -Infinity,
+        counts: new Array(this._bucketCount).fill(0),
+        total: 0
+      })
+    }
+    return this._metrics.get(metric)
+  }
+
+  observe (metric, value) {
+    assertNonEmpty(metric, 'metric')
+    if (typeof value !== 'number' || Number.isNaN(value)) {
+      throw new Error('value must be a number')
+    }
+    const h = this._ensure(metric)
+    h.total++
+    if (value < h.min) h.min = value
+    if (value > h.max) h.max = value
+    if (h.min === h.max) {
+      h.counts[0]++
+    } else {
+      const idx = bucketIndex(value, h.min, h.max, this._bucketCount)
+      h.counts[idx]++
+    }
+    this._stats.observed++
+    this._gossip({
+      type: 'histogram-buckets',
+      metric,
+      min: h.min,
+      max: h.max,
+      counts: [...h.counts],
+      peer: this.peerHex,
+      at: Date.now()
+    })
+    this.emit('observe', { metric, value })
     return true
   }
 
-  summarize () {
-    if (!this._samples.length) return null
-    const vals = this._samples.map((s) => s.value)
-    const sum = vals.reduce((a, b) => a + b, 0)
-    return { count: vals.length, min: Math.min(...vals), max: Math.max(...vals), avg: sum / vals.length }
+  percentile (metric, p) {
+    assertNonEmpty(metric, 'metric')
+    const h = this._metrics.get(metric)
+    if (!h || !h.total) return null
+    const pct = Math.max(0, Math.min(100, Number(p)))
+    const target = Math.ceil((pct / 100) * h.total)
+    let seen = 0
+    for (let i = 0; i < h.counts.length; i++) {
+      seen += h.counts[i]
+      if (seen >= target) {
+        if (h.min === h.max) return h.min
+        const span = (h.max - h.min) / h.counts.length
+        return h.min + span * (i + 0.5)
+      }
+    }
+    return h.max
   }
 
-  _onGossip (d) {
-    if (!d || d.type !== 'histogram-gossip-sync') return
+  snapshot (metric) {
+    const h = this._metrics.get(metric)
+    if (!h) return null
+    return { metric, min: h.min, max: h.max, counts: [...h.counts], total: h.total }
+  }
+
+  metrics () {
+    return [...this._metrics.keys()].sort()
+  }
+
+  _mergeBuckets (local, remote) {
+    if (!remote || !remote.counts) return
+    if (remote.min < local.min) local.min = remote.min
+    if (remote.max > local.max) local.max = remote.max
+    const len = Math.min(local.counts.length, remote.counts.length)
+    for (let i = 0; i < len; i++) local.counts[i] += remote.counts[i]
+    local.total += remote.counts.reduce((a, b) => a + b, 0)
+  }
+
+  _gossip (data) {
+    if (!this._peerMsgs) return
+    gossipSend(this, data)
+    this._stats.gossipOut++
+  }
+
+  _onGossip (data) {
+    if (!data || data.type !== 'histogram-buckets' || !data.metric) return
     this._stats.gossipIn++
-    if (typeof d.value === 'number') this._samples.push({ value: d.value, at: Date.now() })
+    const h = this._ensure(data.metric)
+    this._mergeBuckets(h, data)
+    this.emit('remote-buckets', { metric: data.metric, peer: data.peer })
   }
 
-  getStats () { return { ...this._stats, protocol: PROTOCOL } }
+  getStats () {
+    return {
+      ...this._stats,
+      metrics: this._metrics.size,
+      protocol: PROTOCOL
+    }
+  }
 
   async ready () {
     if (this.swarm || !this.topic) return this
-    await attachGossip(this, { keyPair: this.keyPair, topic: this.topic, protocol: PROTOCOL, onmessage: (d) => this._onGossip(d) })
+    await initModuleSwarm(this, {
+      keyPair: this.keyPair,
+      topic: this.topic,
+      protocol: PROTOCOL,
+      onmessage: (data) => this._onGossip(data)
+    })
     return this
   }
 
   async close () {
+    this._metrics.clear()
     if (this.swarm) await this.swarm.destroy().catch(() => {})
     this.swarm = null
+    this.emit('closed')
   }
 }