level

A level records a quantity that stays where you put it. Every window from a write onwards carries that value, including the windows nobody wrote to, so a chart of it draws a line rather than a row of dots.

import { level, oneOf } from 'metrichouse/core'

export const queueDepth = level('queue_depth', {
  dims: { queue: oneOf(['email', 'export', 'webhooks']) },
  resolution: '1m',
  flush: '1m',
  write: async (rows) => clickhouse.insert({ table: 'queue_depth', values: rows }),
})

queueDepth.set(42, { queue: 'email' })

Import	import { level } from 'metrichouse/core'
Answers	What is this value now, and what was it in between
Storage	One held value per series, carried into every closed window
Row	{ id, bucket_ts, ...dims, value }
Write with	set(), inc(), dec()
Read with	current(), totals(), snapshot()
Use it for	Queue depth, requests in flight, connections checked out, workers running

Every series produces a row in every window, forever, whether or not anything moved. What it costs puts a number on that.

level()

level<D>(name: string, config: LevelConfig<D>): Level<D>

Declares a level. It is inert until a house binds it, and writing to an unbound level throws.

Parameter	Type	Required	Meaning
name	string	yes	The metric name
config.dims	shape	no	Labels to break the value down by
config.resolution	duration	yes	How wide one window is
config.flush	duration	no	The fastest this may ship
config.grace	duration	no	How long a window waits for writes on their way
config.holdFor	duration	no	How long a quiet series keeps reporting
config.value	field type	no	Whether fractions are allowed
config.collect	function	no	Read a value held elsewhere before each window closes
config.collectLead	duration	no	How long before the end of a window to call it
config.collectScope	'fleet' or 'process'	no	Which processes call it
config.write	function	yes	Where the rows go

name

level('queue_depth', { ... })

A non empty string, unique inside a house.

dims

dims?: Record<string, FieldType>      // default: none

The labels this quantity is broken down by. Each combination holds its own value and produces its own row in every window.

dims: { queue: oneOf(['email', 'export', 'webhooks']) }

Dims cost more here than on any other type, because every series reports forever. A dim whose values come and go wants holdFor. A dim cannot be named id, bucket_ts or value, the columns a level writes on every row itself. dims covers the argument itself. A series held from before a dim was added at the end ships the windows it was written in and is carried no further, so current(), totals() and the windows a snapshot carries leave it out.

resolution

resolution: DurationInput      // required

How wide one window is, and therefore how many rows a single series writes per day. One series at '1m' is 1,440 rows a day. At '1s' it is 86,400.

Buckets and time covers the choice, and Durations the format.

flush

flush?: DurationInput      // default: the house default

The fastest this level may ship. resolution has to divide it evenly.

Flushing does more for a level than for the other types: the carry happens here, so a level that never flushes never fills in the windows between writes. See How carry works.

grace

grace?: DurationInput      // default: the house default, then '2s'

How long a window waits after it ends before a flush may claim it, so writes made inside it have time to reach storage. Identical to the counter's.

holdFor

holdFor?: DurationInput      // default: forever

How long a series keeps reporting after its last write. Without it a series holds its value for as long as the process runs, which is the point of a level and the wrong answer for a series that can go away.

export const workerQueue = level('worker_queue', {
  dims: { worker: str() },
  resolution: '1m',
  flush: '1m',

  // A worker that dies stops reporting after five idle minutes, instead of
  // leaving its last queue depth on the chart forever.
  holdFor: '5m',

  write: toClickHouse('worker_queue'),
})

Past holdFor the series is forgotten and produces no more rows. Writing to it again brings it back, starting from that write. A write that lands while a flush is deciding to forget the series wins: the flush checks again at the moment it drops a series, and keeps any series written since it looked.

A series is forgotten from the first window past its hold, whether or not a flush has removed it from storage yet. The level sends its holdFor with every write, so a write to a series past it that is still in storage starts it over the same way: an inc() starts from zero rather than from the value it expired at. The next flush still ships the windows it reported before it expired, then nothing for the windows in between, then the new write onwards. The rows are the same whether a flush ran during the gap or not. An inc() that reaches storage late, aimed at a window inside that gap, starts from zero as well: the hold is measured from the newest write before the window the inc() lands in, not from the write that revived the series.

The clock runs from the window the last write landed in, so it rounds to whole windows rather than to the millisecond. The last window a series reports is the one that holdFor after its last write falls in. With resolution: '1m' and holdFor: '90s', a series written in the 00:00 window reports 00:00 and 00:01, and the same two rows ship whether you flush every minute or once an hour later. holdFor has to be at least one resolution long, and a shorter one throws at declaration, because it would drop a series before the window it was written in had closed. A holdFor so long that it ends past the largest safe timestamp, about 285,000 years from 1970, is the same as holding forever.

Without holdFor, a dim whose values come and go grows without bound: every worker id that has ever appeared keeps writing a row every window. That is the one way a level quietly becomes expensive.

value

value?: FieldType<number, false>      // default: float()

Whether writes accept fractions. A level is fractional by default, which is the opposite of a counter, because the quantities a level holds are often measured rather than counted.

const inFlight = level('requests_in_flight', {
  value: int(),        // whole requests
  resolution: '10s',
  flush: '1m',
  write,
})

A level declared value: int() stops at 9007199254740991, Number.MAX_SAFE_INTEGER, as an integer counter does. A write whose value is past it throws at the call, and an inc() that would move the series past it is refused by the driver and reported to onError.

collect

collect?: (metric: Level<D>) => void | Promise<void>      // default: none

A function MetricHouse calls a little before each window closes, for a quantity that is kept somewhere else and only needs reading. It receives the level, calls set() on it, and the value lands in the window that is about to close and is carried from there like any other write.

const poolInUse = level('db_pool_in_use', {
  value: int(),
  resolution: '1m',
  flush: '1m',
  collect: (self) => self.set(pool.numUsed()),
  collectScope: 'process',     // every process has a pool of its own
  write,
})

The production shape reads a number your application already keeps. Here each job queue is a Redis list in your own Redis, and its length is the depth.

// metrics/schema.ts
import { Redis } from 'ioredis'
import { int, level, oneOf } from 'metrichouse/core'
import { toClickHouse } from './sinks.js'

const redis = new Redis(process.env.REDIS_URL!)
const QUEUES = ['email', 'export', 'webhooks'] as const

export const queueDepth = level('queue_depth', {
  dims: { queue: oneOf(QUEUES) },
  value: int(),
  resolution: '1m',
  flush: '1m',

  // One LLEN per queue, sent together, a second before each minute ends. The
  // default scope, 'fleet', means one server reads them for the whole fleet,
  // because every server would see the same lists.
  collect: async (self) => {
    const lengths = await Promise.all(QUEUES.map((queue) => redis.llen(`jobs:${queue}`)))
    QUEUES.forEach((queue, i) => self.set(lengths[i] ?? 0, { queue }))
  },

  write: toClickHouse('queue_depth'),
})

Use set() inside collect, which states what the quantity is now. inc() and dec() move a series by an amount, and a read of an outside value has no amount to move by.

With house.start(), a timer calls it collectLead before each window ends. Without the scheduler, every flush() calls it first, at most once per window, and house.stop() calls it once more before its final flush. Collecting before a window closes covers each of those, and what happens when it throws, runs long or runs on several servers.

A value lands in the window open at the moment set() is called. A collect that awaits a read has to finish inside the lead, or its value falls into the next window. A collect that is not a function throws at declaration.

collectLead

collectLead?: DurationInput      // default: '1s', or a tenth of a shorter resolution

How long before the end of each window the scheduler calls collect. Make it longer than the read inside collect takes. The default is one second, or a tenth of the resolution when that is shorter. A lead of zero, or one as long as the resolution or longer, throws at declaration, and so does a collectLead given without collect. Identical to the gauge's.

collectScope

collectScope?: 'fleet' | 'process'      // default: 'fleet'

Which processes call collect when several share one driver.

Scope	Calls per window	What the window holds
'fleet'	One, from whichever process asks first	The value that process read
'process'	One per process	The value the last process to write set

Keep 'fleet' for a quantity every process would read the same, such as the lists above. Use 'process' for one each process has its own of, such as its connection pool, and give the level an instance dim so each process holds a series of its own. Without that dim, every process sets the same series and the last one to write wins. On a driver that is not shared, such as memory(), the two are the same. A collectScope given without collect throws, and so does one that is neither of the two. Collecting across a fleet explains how one process is picked.

write

write: (rows: LevelRow<D>[], context: WriteContext) => Promise<void> | void

Where the rows go. Required.

Parameter	Type	Meaning
rows	LevelRow<D>[]	One row per series per window, with value as the series stood when the window closed
context	WriteContext	Which metric, which window, how many, and which attempt

context.total for a level is where every series stood at the end of the batch, rather than the sum of every row in it. A level that sat at 42 for five windows would otherwise report 210, which is a number nothing corresponds to. Writing a sink covers the rest of the contract.

level.set()

set(value: number, dims: Dims): void
set(value: number): void      // no dims declared

Puts the series at this value. It stays there until something changes it.

Parameter	Type	Required	Meaning
value	number	yes	What the series is now
dims	the declared shape	once any dim is declared	Which series to move

queueDepth.set(42, { queue: 'email' })     // it is now 42
queueDepth.set(38, { queue: 'email' })     // it is now 38

A second write in the same window replaces the first. On a gauge the same two calls would be two observations that fold together, and that difference is what separates the two types.

Readings are ordered by the window they were written in, not by the moment they reached storage. A set() for an earlier window than the series' newest write fills its own window and the empty ones after it, and leaves the newer value current. That is what keeps two processes writing across a window boundary right, and it has a cost when a server's clock steps backwards. A reading taken after the step lands in an earlier window than the reading taken before it, so the older reading stays current. current() keeps returning it, and the carry keeps repeating it, until something writes to its window or a later one. Buckets and time covers the same effect from the clock's side.

Returns nothing, and returns before storage has acknowledged anything.

Throws immediately on an unbound level, a value that is not finite, a fraction or a whole number past 9007199254740991 on a level declared value: int(), or dims that are missing, unknown or ill typed. inc() and dec() also throw for a first argument that is neither a number nor a dims object, such as a bigint, which TypeScript accepts on a level with no dims.

A value too large for a JavaScript number cannot reach storage either. An inc() that would move a series past about 1.8e308 is refused by the driver, and because inc() has already returned by then, the error goes to the house's onError. The series keeps the value it had.

level.inc()

inc(delta: number, dims: Dims): void
inc(dims: Dims): void
inc(delta: number): void      // no dims declared
inc(): void                   // no dims declared

Moves the series up by delta, or by 1. For quantities that are counted in and out rather than measured.

const inFlight = level('requests_in_flight', { resolution: '10s', flush: '1m', write })

app.use(async (c, next) => {
  inFlight.inc()
  try {
    await next()
  } finally {
    inFlight.dec()
  }
})

A series nothing has written to starts at zero when it is moved, so the first inc() puts it at 1.

level.dec()

dec(delta: number, dims: Dims): void
dec(dims: Dims): void
dec(delta: number): void      // no dims declared
dec(): void                   // no dims declared

Moves the series down by delta, or by 1. The mirror of inc(), and nothing stops a series going negative, because a pairing that runs one way more often than the other is worth seeing rather than hiding.

level.current()

current(dims: Dims): Promise<number | undefined>
current(): Promise<number | undefined>      // no dims declared

What one series is at right now.

await queueDepth.current({ queue: 'email' })
// 42, or undefined if nothing has ever written to it

Returns the held value, read from the series rather than from the open window. The window you are in may well be empty, and the level is still 42. That is the difference from gauge.current().

undefined rather than 0 for a series nothing has written to. A zero claims the queue exists and is empty, which is a different thing from not knowing yet. A series past its holdFor also reads undefined, from the first window after its last one, even before a flush has removed it from storage. totals() leaves it out the same way.

Only the one series is read, on a driver that implements readLevel, as both built in drivers do. totals() reads every series, because it adds them all.

level.totals()

totals(): Promise<number | undefined>

Every series added together.

await queueDepth.totals()
// 61, the depth across every queue

Returns the sum of the held values, or undefined when no series has ever been written to.

On a level declared value: int(), the series are added as whole numbers, exactly, and a total past 9007199254740991 on either side rejects rather than returning a nearby whole number. So does a snapshot() that adds series together past it. Each series stays below it on its own, but several added together can pass it. On a fractional level, a total past the largest number a double holds, about 1.8e308, rejects the same way rather than returning Infinity.

A level declared value: int() over series a float() level stored can hold fractions. Then the series are added as doubles instead, so 1.5 and 0.5 total 2, and a total that is not a whole number rejects with an error that says a stored value is a fraction.

Adding is the merge a level can make honestly, because every held value is true at the same moment. A gauge drops last from its totals for the opposite reason.

level.snapshot()

snapshot<O extends SnapshotOptions>(options?: O): Promise<LevelLiveRow<D, O>[]>

Every unflushed window, as rows. That includes the windows the next flush will carry a held value into, not only the ones something wrote to, so a queue that sat at 42 for five minutes reads as five rows of 42, with the same ids those rows will ship under. With complete: false, the open window is included too, at the value the series is at right now. Nothing past the open window is returned, even when to reaches into the future, because a window that has not started has no value yet.

await queueDepth.snapshot()
await queueDepth.snapshot({ dims: { queue: 'email' } })
await queueDepth.snapshot({ rollup: 'sum', groupBy: ['queue'] })

A rollup takes the latest value per series, then adds the series up, in that order. Merging one series across windows gives the latest of them, because the earlier values have been superseded. Merging several series inside one window gives their sum, because all of them are true at once.

Every option is in Snapshot options.

level.flush()

flush(options?: FlushOptions): Promise<MetricFlushReport>

Carries every series forward into the closed windows that have none, then ships them. How carry works covers the first half, and Flush options the argument and the report.

level.drain()

drain(): Promise<void>

Resolves once every write issued before the call has reached the driver. On a level declared with collect, it first waits for a collect still running, so the value it is about to write is waited for too. Draining does not carry or claim anything. Under immediate delivery it waits for the send to write that follows each write as well.

level.rowShape()

rowShape(): RowShape

queueDepth.rowShape().columns.map((c) => c.name)
// ['id', 'bucket_ts', 'queue', 'value']

Properties

Property	Type	Value
name	string	The name it was declared with
kind	'level'
storage	'bucketed'	It holds one value per series and fills windows from it
dims	Shape	The declared dims
resolutionMs	number	resolution, parsed
flushMs	number	flush, parsed, including one taken from the house. Throws on an unbound metric that declared no flush
graceMs	number	grace, parsed, including one taken from the house. 2000 when neither sets it
holdForMs	number | undefined	holdFor, parsed. undefined when a series holds forever
isFloat	boolean	true unless value: int() was declared
isBound	boolean	true once a house has registered it
write	WriteFn	The function it was declared with

How carry works

Each flush does one thing before it claims anything. It walks every series forward from the last window it carried, up to the newest closed window, and writes the held value into each window that has none.

writes        42 .  .  .  38 .  .
stored        42 42 42 42 38 38 38
              ^           ^
              set()       set()

The walk goes forwards through what was actually written, rather than stamping the current value across the whole gap. Set a queue to 42 at noon and to 38 at three, and the windows in between hold 42. The queue changed at three, so it did not change at noon.

A value somebody wrote always beats a carried one, whichever of the two lands first. Two processes carrying the same window write the same number, so they cannot disagree.

Nine consequences worth knowing:

• A series appears from its first write. Nothing is backfilled before it, so a queue declared on Monday and first written on Friday has no Monday rows.
• The carry happens at flush. A level that never flushes never carries, so the windows appear when the metric ships rather than as the clock passes.
• Coming back from downtime leaves a gap. A process off for a day owes 86,400 windows per series at resolution: '1s', and writing them all would claim the queue was measured throughout a period when nothing was watching. Past MAX_CARRY_BUCKETS, which is 10,000, the older windows are skipped and the gap stays in the data. A write inside the skipped stretch still counts: the first window after the gap carries the newest value written before it, not the value from before the gap began. The 10,000 are counted back from the newest window a flush may claim, and snapshot() counts from the same place, so a live read and the next flush skip the same windows.
• The window a series was written in ends at its last write. Set a queue to 5 and then to 3 inside one window, or inc(5) then dec(2), and that window and every empty one after it hold 3.
• Order does not matter only when every write to a series is inc() or dec(), and none lands past holdFor. When one server's inc() for a window reaches Redis after another server's inc() for the next window, the late one still counts in its own window and in every window after it. A series that also takes set() can end up different depending on which write arrives first. A late inc() adds its change to a later window a set() wrote. A late set() fills its own window and the carried windows after it, and does not replace a value written for a later one, so a later window an inc() wrote is not rebased onto it. An inc() that lands more than holdFor after the series' newest write starts from zero.
• A series begins at its earliest write, whichever arrives first. When the first write to reach storage is for a later window than a write that arrives after it, the series begins at the earlier one, and every window between the two ships. An inc() in the earlier window starts from zero.
• A flush sees a write that lands while it runs. Each empty window is filled with the value in effect just before it at the moment it is filled, so a set() that reaches storage after the flush read the series still reaches every empty window after it. A set() that lands after the flush has filled the windows after it replaces those carried values, up to the next window somebody wrote to.
• A late write that missed its window moves forward. A write aimed at a window that has already shipped lands in the first window of its own resolution at or past the watermark, as on every type. A set() moved that way is dropped when that window already holds a reading taken in that window's own time, because that reading was taken later. When the window holds only a value an earlier late write moved there, both missed their windows, and the set() that arrives second replaces it.
• A series that starts late during a flush can lose windows. A flush reads every series, fills the empty windows, and then claims. When a series' first write, or a write that moves its start earlier, reaches storage between that read and that claim and is stamped below the watermark, the flush claims those windows without a value in them, and they never ship. It takes a write stamped more than grace in the past, so a clock that runs behind by more than grace is the usual cause. Keep clocks in sync, or raise grace, to keep it away.

What it costs

A counter writes a row for a window something happened in. A level writes one for every window, for every series, forever.

Series	Resolution	Rows per year
1	'1m'	525,600
10	'1m'	5,256,000
10	'10s'	31,536,000
100	'1m'	52,560,000

Those numbers arrive whether or not anything moved. Two settings bring them down: a wider resolution, and holdFor on a dim whose values come and go.

The same three writes, stored two ways.

Reach for a gauge when you are sampling something, and for a level when the gap between writes is the part you need filled.

The row

{
  id: '...',
  bucket_ts: Date,
  queue: 'email',
  value: 42,
}

Column	Meaning
id	Derived from the name, the window and the dim values
bucket_ts	The start of the window
one per dim	The label values for this series
value	What the series was at when the window closed

Inside write, each row is a LevelRow<D>. See Rows are typed.

Table schema

ClickHouse

CREATE TABLE queue_depth (
  id         String,
  bucket_ts  DateTime64(3),
  queue      LowCardinality(String),
  value      Float64
)
ENGINE = ReplacingMergeTree
ORDER BY (bucket_ts, queue);

Postgres

CREATE TABLE queue_depth (
  id         TEXT PRIMARY KEY,
  bucket_ts  TIMESTAMPTZ NOT NULL,
  queue      TEXT NOT NULL,
  value      DOUBLE PRECISION NOT NULL
);

Queries

-- The deepest each queue got per hour, and where it sat on average.
SELECT
  toStartOfHour(bucket_ts) AS hour,
  queue,
  max(value) AS deepest,
  avg(value) AS typical
FROM queue_depth
WHERE bucket_ts >= now() - INTERVAL 1 DAY
GROUP BY hour, queue
ORDER BY deepest DESC;

What a level does not do

It does not aggregate inside a window. A series written five times in one minute stores the fifth value, and the other four are gone. For the minimum and maximum inside each window, use a gauge, and writing to both is reasonable.

It does not know why it changed. A level going from 42 to 38 does not say whether four items were processed or four were cancelled. Pair it with a counter when the reason matters.

Patterns

A job queue, reported by its worker

// metrics/schema.ts
import { level, oneOf, str } from 'metrichouse/core'
import { toClickHouse } from './sinks.js'

export const queueDepth = level('queue_depth', {
  dims: {
    queue: oneOf(['email', 'export', 'webhooks']),
    instance: str(),
  },

  // One row per minute per queue per instance. Coarse enough to stay cheap,
  // fine enough to show a backlog building before anyone complains.
  resolution: '1m',
  flush: '1m',

  // An instance that stops reporting for ten minutes has gone away, and its
  // last depth should go with it rather than sit on the chart.
  holdFor: '10m',

  write: toClickHouse('queue_depth'),
})

// queue/report.ts
import { queueDepth } from '../metrics/schema.js'
import { queues } from './queues.js'

const instance = process.env.HOSTNAME ?? 'local'

export function startReporting() {
  // Written on change rather than on a timer: a level does not need a sample
  // per window, because the windows in between are filled in for it.
  for (const [name, queue] of Object.entries(queues)) {
    queue.on('change', () => {
      queueDepth.set(queue.size(), { queue: name, instance })
    })
  }
}

Pausing intake on depth

The held value is readable before anything ships, so a level can drive a decision.

export async function shouldPause(queue: string) {
  const depth = await queueDepth.current({ queue, instance })

  // Nothing has reported yet, so there is no backlog to react to.
  if (depth === undefined) return false

  return depth > 10_000
}

Playground

A level buckets like a counter, so the same two settings decide the same things. What changes is that every series produces a row in every window rather than only in the windows it was written in.

Try it: resolution inside flush

resolution1mhow wide one bucket isflush1mthe fastest this metric may shipgrace2show long a late write still landslabel combinations4one row each, per bucket

Each outer box is one flush. Every bucket inside it ships together, in one call to your write function.

1buckets per flushrows arrive together

4rows per flushbuckets × combinations

5.8krows per daywhat your table stores

1.4kwrites per daycalls to your database

The declaration these settings produce

level('queue_depth', {
  resolution: '1m',
  flush: '1m',
  write: async (rows) => db.insert(rows),
})

Your sink receives id, bucket_ts, …dims, value, one row per combination per bucket.