How to Collect Metrics from a Fleet of Databases

This is not a web scraper, despite the name. scrape here means scrape metrics: a monitoring collector — a program that runs in the background — pulls numbers out of hundreds of Oracle, MySQL, PolarDB and PostgreSQL databases, plus Linux, Solaris and AIX hosts, on a timer, and ships them to Riemann. Riemann is a time-series database: a store that keeps each measurement as a timestamped event so you can graph it later.

What this post is about. It is about the Kotlin program that does this — nirvana-scrape — and about how its pieces fit together: how a piece of work is described, how it is run, how a new kind of database is added, and how one slow instance is kept from stalling the other 199.

What you need to know first. Four words. A task is one piece of work the collector can run. A driver is the plugin that knows how to talk to one kind of target. A pipeline is an ordered list of stages that a task passes through. A schedule is the timer that decides when to collect. Everything else is explained where it appears.

The project lives in the Nirvana product family (gitlab.enmotech.com/nirvana/); the published artifact is com.enmotech.bethunepro, the collector that ships inside Bethune Pro, a commercial database-monitoring product. The name nirvana-scrape is the repository. The thing you would actually deploy is a daemon — a program that runs in the background and never stops.

The situation

A DBA — a database administrator, the person who keeps the databases healthy — opens a dashboard and expects to see, for a fleet of 200 databases: session counts, wait-time histograms, table-space usage, replication lag (how far a standby copy trails its primary). Somebody has to go and ask each instance, every thirty seconds, and put the answers somewhere queryable.

The difficulty is not the SQL. It is that every target is different in four ways at once:

  • Transport: JDBC for databases (the standard Java library for opening database connections and running SQL), SSH for hosts (the encrypted remote-login protocol for running commands on another machine), with a Perl interpreter in between for the host scripts.
  • Dialect: Oracle’s DBA_ views — its built-in metadata tables — are not MySQL’s information_schema, and Data Guard (Oracle’s built-in primary/standby replication) needs its own verbs entirely.
  • Failure mode: a database does not fail like a host does. A host may refuse the SSH handshake. An Oracle instance may accept the connection and then hang for two minutes inside a GROUP BY.
  • Blast radius: one slow instance must never delay the other 199.

There is a second, less obvious requirement. The same collector also has to serve ad-hoc work: an operator opens a query console, types SQL against one of those 200 instances, and expects an answer in seconds — that query is a 2000-row GROUP BY on a production box during business hours. It must not starve the scheduled metrics collection on the same machine; “starve” means delaying it so long the data goes stale.

How the collector is put together

A task is a record, not a class

The central abstraction is not a Task interface with execute()/onError() methods. A task is a record — a plain data holder with no behaviour of its own:

1
2
3
4
5
6
data class ScrapeTask(
    val id: String = guid(),
    var action: String? = null,
    @JvmField var scripts: ScrapeScripts = ScrapeScripts(),
    ...
)

data class is Kotlin’s shorthand for a class whose only job is to hold values; the compiler writes the equals, hashCode and toString methods for you. The file is nirvana-scrape-core/src/main/kotlin/com/enmotech/bethunepro/scrape/core/domain/ScrapeTask.kt:17. Each field is a slot you can fill in without writing any Kotlin. action is a verb — ORACLE_QUERY, LOG_FILE_READ, SHELL_TEXT_LINE, MYSQL_QUERY. scripts holds the code to run on each platform. The ... stands for the fields left out: depends, which lists other tasks this one needs, and two parameter objects carrying typed input and output values. The rest is bookkeeping: a key() helper that gives the task a globally unique name, a newInstance() copy method, and a builder.

This is what makes “custom task” work. An operator adds a new monitoring item as a JSON document — pick an action, attach a script per platform, give it a cron expression (a short scheduling pattern, like */30 * * * * for “every 30 seconds”), a timeout, and a set of parameters. The result is a new metric with no code and no rebuild. CUSTOM_TASK.md is the manual for that; it is the feature the whole design is arranged around.

A pipeline: an ordered list of stages

Behaviour is attached to a task through named phases rather than by subclassing. There are three pipelines — one for connecting, one for the HTTP request, one for collecting — and the collect one is where the work happens. Its phase list, in order, is Before, Transform, Render, Valid, Collecting, Collected, After, declared as ScrapeCollectPipeline : Pipeline<ScrapeResponseData, ScrapeCollectCall> in nirvana-scrape-core/.../core/request/ScrapeRequestPipeline.kt:49.

Reading that list left to right gives you the life of a collection: prepare it, work out the parameters, choose the script, check it is legal, run it, tidy up, finish. A third-party plugin can splice its own render phase into the chain without touching core. The pipeline itself is a hand-rolled interceptor chain (common/Pipelines.kt) — an interceptor is a small step that wraps every request — with Pipeline.merge() for splicing.

The pipeline even reimplements coroutine interception to avoid allocating a CoroutineContext per interceptor. A CoroutineContext is the object that carries “who is running this” — which thread, which cancellation scope — and building one per interceptor per task adds up. That micro-optimisation is not one I would defend in a new project. It exists because this code runs once per instance per interval, on a fleet of hundreds, forever.

Plugins are the drivers

The extension point — the place where outside code can plug into the program — is pf4j, a plugin framework for the JVM, the Java virtual machine that runs Kotlin programs. A plugin is a driver:

1
2
3
4
5
interface ScrapeDriver {
    fun id(): String
    fun createConnection(instance: ScrapeInstance): ScrapeConnection?
    fun isSupportAction(action: String, instance: ScrapeInstance): Boolean
}

nirvana-scrape-core/.../core/spi/ScrapeDriver.kt:18. An interface is a list of methods that an implementor has to provide. createConnection opens a connection to one instance — the ? says it may return nothing at all. isSupportAction answers “do you handle this kind of task?”. The base class makes the link explicit: ScrapePlugin extends pf4j’s Plugin and implements ScrapeDriver at the same time. Its id() reads the plugin id out of pf4j’s descriptor, falling back to the literal string "plugin".

Plugins are discovered through pf4j’s extensions.idx file, generated by kapt (a compile-time code generator for Kotlin). On the RegisterEvents.Started event the framework starts the loaded plugins, keeps the ones that are ScrapePlugins, and hands each to the driver manager.

The collector is stateless

There is no database, no SQLite, no local task table. The monitoring backend (Riemann) is the source of truth for what to collect. The collector is stateless — it keeps no memory of past work between runs — so it is a fire-and-forget agent that can be killed and restarted at any moment. The only local state is scrape.conf and the driver configs.

Two connection pools, on purpose

This one surprised me when I read the changelog. There are two pools. Monitoring items use a persistent pool. Drill-down and inspection queries use a temporary pool. The reason is stated in the 1.2.3 entry: a long ad-hoc query must not be able to starve the scheduled metrics.

1
2
3
4
5
connectData.connection = if (!instance.isTemp) {
    connectionContainer[instance.connectKey()]?.getConnection(connectCall.tag())
} else {
    context.collectCall.takeTempConnection(instance)
}

nirvana-scrape-core/.../core/feature/ScrapeConnectionManager.kt:108. The if is the whole idea. isTemp means “this connection is for a one-off query”. One-off queries get a connection of their own that is thrown away afterwards, so they cannot sit on top of a connection a scheduled metric needs. connectKey() is the pool key — the identity of the instance being talked to.

How it works

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
┌──────────────┐  1s ticker   ┌────────────────┐  cron match
│  Schedule    │─────────────►│ per-instance   │─────────────┐
│  Service     │              │ bounded chan   │             │
└──────────────┘              └────────────────┘             │
        ▲                                                    ▼
        │                                            ┌──────────────────┐
        │                                            │ ScrapeCollector  │
        │                                            │  beforeHandle    │
        │                                            └────────┬─────────┘
        │                                                     │
        │              ┌──────────────────────────────────────┴───────┐
        │              │ depends: fan out (async) → awaitAll → merge  │
        │              └──────────────────────────────────────────────┘
        │                                                     │
        │                                            ┌────────▼─────────┐
        │                                            │ Connect → Valid   │
        │                                            │ → Collect → Render│
        │                                            └────────┬─────────┘
        │                                                     │
        │                                            ┌────────▼─────────┐
        └──────────────── Riemann events ◄────────────┤  Send            │
                                                      └──────────────────┘

The boxes are components and the arrows show the order work flows through them. The schedule service wakes once a second and checks every task’s cron expression; when one matches, the task is pushed into a bounded queue — a queue with a fixed capacity, so a backlog cannot grow without limit. The collector picks tasks off that queue and, before collecting anything, beforeHandle runs the task’s dependencies: each one is launched concurrently with async, all are waited for together with awaitAll, and the results are merged. Only then does the pipeline itself run Connect → Valid → Collect → Render, and the finished values are sent on as Riemann events. The arrow on the far left curving back up is the cycle restarting: the collector keeps going forever.

Two independent layers of parallelism. The engine fans out over tasks and instances:

1
2
3
4
5
6
suspend fun transformTask(register: ScrapeRegister, requestCall: ScrapeRequestCall) =
    requestCall.tasks.map {
        async(createCallContext() + requestCall.coroutineContext) {
            transformInstance(register, requestCall, it)
        }
    }.awaitAll().merge(requestCall)

nirvana-scrape-core/.../core/ScrapeEngine.kt:85. map turns the list of tasks into a list of coroutines — in Kotlin, a coroutine is a piece of work that can pause and resume instead of blocking a thread. async starts each one without waiting; awaitAll collects the results. Both come from Kotlin’s coroutines library — the standard way to say “start several things in parallel and wait for all of them”.

Dependencies are a directed acyclic graph — a set of tasks pointing at other tasks, with no loops — joined with the same async/awaitAll pair. There is no explicit “start node”: the requested task is the root, and its dependencies are resolved recursively through the same pipeline. That is why a metric can depend on another metric without anyone declaring an order.

Failure containment

The rule that matters in a fleet collector is that one bad instance cannot take down the request. The collection step is wrapped end to end:

1
2
3
4
5
return try {
    val connection = findInstances(context, register)?.connection
        ?: return context.defaultValue().addThrows(...)
    connection.process(action, context)
} catch (e: Exception) { context.defaultValue().addThrows(e) }

nirvana-scrape-core/.../core/feature/ScrapeCollector.kt:287. The ?: marks are Kotlin’s shorthand for “if this is null, stop here and give back this other value”. defaultValue() is an empty result that can carry an attached error. So a failure becomes a value — an error row attached to the response — not an exception that unwinds the fan-out. If one instance throws, the other 199 still finish and still report. The original spells out the full connection-error message on the truncated line, and it also returns the empty result when the task has no matching script.

A guard also runs before anything touches the wire, so a misconfigured instance fails with a clear message instead of a connection error 30 seconds later. The Valid phase refuses a task with an empty action. It also looks up the driver for the instance’s driver and version before starting, throwing a clear “can’t find driver” error if there is none.

Trade-offs

Blocking I/O inside coroutines. connection.process(action, context) is a plain blocking JDBC or SSH call inside a suspend function. The coroutines here buy structured concurrency — every spawned job belongs to a scope that tracks, cancels and waits for it — and not non-blocking I/O. The engine backs them with a dedicated ThreadPoolExecutor, a pool of worker threads sized from configuration.

The TestEngine instead uses Dispatchers.Default, and the README admits the downside in Chinese: “灵活性偏低” — less flexible, because you cannot size it at runtime.

Timeouts are detected, not enforced. The scheduler compares response and request timestamps and throws a TimeoutException after the fact. Nothing cancels the in-flight query. There is a ThreadPoolWithTimeoutExecutor in the tree — a DelayQueue-based supervisor that interrupts timed-out threads — but it is commented out. The reason is sound: interrupting a JDBC call does not reliably interrupt the database, so a hard timeout would leave a worker stuck in a socket read forever. Detect the timeout, report it, and let the connection be reclaimed later.

SSH throughput was the real bottleneck, and the fix was a hack. The 1.2.3 changelog entry reads, roughly: optimise SSH stream reading by pausing 300 ms on each InputStream read, so that a single Scrape can monitor 100 Oracle instances on a 6-core host. That is a polling workaround — the reader sleeps, then checks a cache. It is the kind of fix that only makes sense once you have measured where the time actually goes.

The JVM’s memory cost is real. The changelog shows the pain: removing the fastjson dependency because it could cause OOM (out-of-memory), replacing CopyOnWriteList with ArrayList in the connection pool, removing locks in favour of lock-free structures. JVM startup time is irrelevant for a daemon. GC — garbage collection, the JVM’s automatic memory cleanup, which pauses every thread while it runs — and heap tuning are not.

The release build skips the tests. build.sh runs gradle build -x test. The test suite is substantial — 132 test files, ~15,600 lines, with PipelinesTest (21 tests on the hand-rolled pipeline) and ScheduleImplTest (22 on the vendored cron parser) — but the artefact that ships is built without them.

Two JSON stacks coexist. The core unified on GSON, while the Riemann wire events still use kotlinx-serialization. A small thing that costs a dependency.

What I would do differently

The timeout story needs to be honest in the API. A task that exceeds its timeout is reported late, and the caller cannot tell “this instance was slow” from “this instance hung”. Exposing the elapsed time next to the deadline, and treating “timed out” as a first-class outcome rather than an exception, would make the report much easier to act on.

The coroutine pipeline is over-engineered. The hand-rolled suspendCoroutineUninterceptedOrReturn saves an allocation per interceptor per task. On a fleet collector that is measurable, but it makes the pipeline the hardest part of the codebase to read and the part most likely to be subtly wrong. Keep the interceptor model, throw away the custom interception.

The SSH fix should have been a streaming fix. A 300 ms sleep per read is a global slowdown dressed as a local optimisation. The right answer is a buffered reader on a dedicated thread per host, with the coroutine awaiting a channel — a queue one task fills and another drains. It is more code, and it would have removed a whole category of performance questions.

And the real one: this is where the Rust rewrite comes from. The Kotlin version works and is well tested, but it ships a 40 MB fat JAR — one archive with the program and all its libraries bundled inside — pays GC pauses on a fleet collector, and runs blocking JDBC and SSH calls through a coroutine pool that was never designed for them. The next post is about the Rust rebuild: never a port, but two implementations born in the same commit.

References

  • pf4j — the plugin framework the drivers are built on
  • Ktor — the HTTP framework the app module is built on
  • Kotlin coroutines — what async, awaitAll and suspend mean
  • sshj — the SSH client the host driver uses to reach monitored machines
  • RFC 4253 — the SSH transport protocol underneath that client
  • Ktor plugins — how the HTTP API installs status pages and content negotiation
  • Ktor routing — how the /monitor, /task and /instance endpoints are declared
  • Ktor status pages — turning an exception into a response instead of a blank 500
  • Ktor serialization — the JSON serialisation used by the query console