Something went wrong. Try again.
Monorepo for Tangled tangled.org
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223export type Labels = Readonly<Record<string, string>>;
interface Attribute { key: string; value: { stringValue: string };}
export interface DataPoint { attributes?: Attribute[]; startTimeUnixNano?: string; timeUnixNano: string; asInt?: string; count?: string; sum?: number; bucketCounts?: string[]; explicitBounds?: readonly number[];}
export interface Metric { name: string; unit: string; sum?: { dataPoints: DataPoint[]; aggregationTemporality: number; isMonotonic: boolean }; histogram?: { dataPoints: DataPoint[]; aggregationTemporality: number }; gauge?: { dataPoints: DataPoint[] };}
interface Series<T> { labels: Labels; value: T;}
const key = (labels: Labels) => JSON.stringify(Object.entries(labels).sort());
const attributes = (labels: Labels): Attribute[] => Object.entries(labels).map(([name, value]) => ({ key: name, value: { stringValue: value } }));
class Family<T> { readonly series = new Map<string, Series<T>>(); constructor(private readonly init: () => T) {} at(labels: Labels): T { const id = key(labels); let entry = this.series.get(id); if (!entry) this.series.set(id, (entry = { labels, value: this.init() })); return entry.value; }}
class Counter { private readonly family = new Family(() => ({ n: 0 })); add(labels: Labels, n = 1) { this.family.at(labels).n += n; } points(start: bigint, now: bigint): DataPoint[] { return [...this.family.series.values()].map(({ labels, value }) => ({ attributes: attributes(labels), startTimeUnixNano: String(start), timeUnixNano: String(now), asInt: String(value.n) })); }}
class Histogram { private readonly family: Family<{ count: number; sum: number; buckets: number[] }>; constructor(private readonly bounds: readonly number[]) { this.family = new Family(() => ({ count: 0, sum: 0, buckets: bounds.map(() => 0).concat(0) })); } record(labels: Labels, value: number) { const point = this.family.at(labels); const i = this.bounds.findIndex((bound) => value <= bound); point.buckets[i === -1 ? this.bounds.length : i]++; point.count++; point.sum += value; } points(start: bigint, now: bigint): DataPoint[] { return [...this.family.series.values()].map(({ labels, value }) => ({ attributes: attributes(labels), startTimeUnixNano: String(start), timeUnixNano: String(now), count: String(value.count), sum: value.sum, bucketCounts: value.buckets.map(String), explicitBounds: this.bounds })); }}
interface Instrument { name: string; unit: string; kind: "counter" | "histogram" | "gauge"; read: (start: bigint, now: bigint) => DataPoint[];}
// the otel sdk is much more than the four instruments here needexport class Registry { private readonly instruments: Instrument[] = [];
counter(name: string, unit: string) { const counter = new Counter(); this.instruments.push({ name, unit, kind: "counter", read: (s, n) => counter.points(s, n) }); return counter; }
histogram(name: string, unit: string, bounds: readonly number[]) { const histogram = new Histogram(bounds); this.instruments.push({ name, unit, kind: "histogram", read: (s, n) => histogram.points(s, n) }); return histogram; }
gauge(name: string, unit: string, read: () => number) { this.instruments.push({ name, unit, kind: "gauge", read: (_, now) => [{ timeUnixNano: String(now), asInt: String(read()) }] }); }
// cumulative because then a push that never arrives is covered by the next one payload(resource: Labels, start: bigint, now: bigint) { const cumulative = 2; return { resourceMetrics: [ { resource: { attributes: attributes(resource) }, scopeMetrics: [ { scope: { name: "svfe" }, metrics: this.instruments.map(({ name, unit, kind, read }): Metric => { const dataPoints = read(start, now); switch (kind) { case "counter": return { name, unit, sum: { dataPoints, aggregationTemporality: cumulative, isMonotonic: true } }; case "histogram": return { name, unit, histogram: { dataPoints, aggregationTemporality: cumulative } }; case "gauge": return { name, unit, gauge: { dataPoints } }; } }) } ] } ] }; }}
const nanos = (ms: number) => BigInt(Math.round(ms)) * 1_000_000n;
export interface PushOptions { registry: Registry; /** base url, /v1/metrics is added */ endpoint: string; /** `user:password`, empty for none */ auth: string; resource: Labels; intervalMs: number; fetch?: (url: string, init: RequestInit) => Promise<Response>;}
export const pushMetrics = ({ registry, endpoint, auth, resource, intervalMs, fetch = globalThis.fetch}: PushOptions) => { const start = nanos(Date.now()); const headers: Record<string, string> = { "content-type": "application/json" }; if (auth) headers.authorization = `Basic ${btoa(auth)}`; let failing = false; const push = async () => { const body = JSON.stringify(registry.payload(resource, start, nanos(Date.now()))); try { const res = await fetch(`${endpoint}/v1/metrics`, { method: "POST", headers, body, signal: AbortSignal.timeout(intervalMs) }); if (!res.ok) throw new Error(`status ${res.status}`); failing = false; } catch (error) { if (!failing) console.warn("[metrics] push failed:", error); failing = true; } }; const timer = setInterval(push, intervalMs); timer.unref?.(); return { push, stop: () => clearInterval(timer) };};