How to Split a Collector into Protocol Services

The Kotlin collector had one process that could talk to a database over JDBC, to a host over SSH, and — thanks to a stack of Perl scripts — to a running Oracle instance on a host it could only reach through a shell. The Rust rewrite had the same three jobs to do and a different problem: the three toolchains do not belong in one binary.

So the Rust version is not one program. It is four, speaking one protocol.

What this post is about. It is about splitting one collector into several small services that talk over a network, and about the single protocol they share. You will learn why the split follows the dependency set rather than the feature list, what the wire looks like, and how a Lua script ends up able to call a remote Java database service as if it were a local function.

What you need to know first. A service here is a separate process that does one job and answers requests over a socket. A schema is the name of a group of remote functions — host, database, agent — and a datasource is one function in that group, together with the types of its arguments. Lua is a small scripting language; mlua is the Rust library that embeds it.

The situation

The collector has to reach three kinds of targets, and each needs a different dependency set that is hostile to the others:

Target Needs Language home
A database (Oracle, MySQL, MSSQL, PostgreSQL, PolarDB) JDBC drivers — ojdbc7.jar, mssql-jdbc, polardb-jdbc Java
A remote host over SSH sshj, key handling, SFTP Java
The machine it is running on heim (CPU/memory), diskus (directory size), notify (log watch) Rust

JDBC is Java’s database API: one jar per database vendor, each with its own quirks. SSH is the remote-shell protocol every server already speaks.

Putting a JVM and the heim crate in the same process is possible and terrible: the JDBC driver set fights the native sysinfo crate for the classpath and the build, and a JVM crash in a JDBC call takes the collector with it.

So the dependency set is the service boundary.

The topology

Four processes, one protocol (“Bee Data Transmission Protocol”, libcore/codec):

Process Language Role Default port
bee (collector) Rust (tokio) Runs the tasks, calls the other services, ships metrics to Riemann server 25205, HTTP 25201
agent Rust (tokio) Native host agent; serves the agent schema; also dials back 25204
host Java (Netty + sshj) SSH remote host; serves the host/dg schema 6141
database Java (Netty + JDBC) The databases; serves the database schema 6142

tokio is Rust’s asynchronous runtime: it runs many sockets on a small number of threads, so a program can wait on thousands of connections without a thread each. Riemann is the time-series database the metrics are shipped to.

Addressing is by configuration, not discovery: configs/proxy.toml maps a schema to a protocol to a list of addresses, and the collector resolves the instance’s host:port through that table. A mysql instance resolves to the database service on 127.0.0.1:6142; an agent instance resolves to the agent’s own address.

The interesting part is that the collector is also a server. It serves the bee schema, and the agent uses it: the agent reads the collector’s address out of the instance parameters, opens a client connection back, fetches the list of log files it should watch, and pushes log bytes home. One binary, both roles of the same protocol.

The wire protocol

The frame is deliberately small:

1
HEAD 0xFF 0xFF | CMD(1) | LEN(4, u32) | DATA | MSG_LEN(4) | END 0x0D 0x0A

Each box is one field of the message: two start bytes, a one-byte command, a four-byte length (u32 means “unsigned 32-bit integer”), the payload, a second length field, and two end bytes. The overhead is a fixed 13 bytes, and MSG_LEN must equal LEN + 13. There are six commands: connect request/response, statement request/response, ping.

The value encoding is five types plus nil — integer, number, boolean, string, bytes — and there is no map or struct on the wire:

If the data you need to transfer is complex, serialize it into a byte stream yourself and send it as Bytes. The protocol does not care.

That escape hatch is what lets one protocol carry a JDBC result set, a directory listing and a log line without a schema registry. The cost is that the schema lives in the payload, and the two implementations (Rust libcore/codec, Java libcore/java) can drift — the protocol document still says “no arrays” while the code has had a sixth type, Array, for a while.

The connect handshake negotiates the signature

1
2
3
4
5
6
7
8
pub async fn connect(addr: &str, instance: Instance) -> Result<SchemaClient> {
    let login_timeout = instance.get_param::<u32>("login_timeout").unwrap_or(10) as u64 + 1;
    let connection_timeout = instance.get_param::<u32>("connection_timeout").unwrap_or(10) as u64 + 1;
    let stream = timeout(Duration::from_secs(connection_timeout), TcpStream::connect(addr)).await...??;
    stream.set_nodelay(true)?;
    let (reader, writer) = stream.into_split();
    let mut reader_framed = FramedRead::new(reader, PacketCodec);
    let mut writer_framed = FramedWrite::new(writer, PacketCodec);

async fn is a Rust function that can wait without blocking its thread; await is where it waits. The two timeouts come from the instance’s parameters, default to 10 seconds, and get + 1 added — the client always gives up a second after the server does. set_nodelay(true) turns off Nagle’s algorithm, which delays small writes to batch them; for a request/response protocol that delay is pure latency. into_split cuts the socket into a reading half and a writing half, and FramedRead/FramedWrite wrap each half in PacketCodec, the object that turns a byte stream into frames and back.

libcore/codec/src/client.rs:55. The connect response carries the datasource signatures:

ConnectResp::Ok { schema, version, datasources: name → Vec<DataType> }

So the client learns, at connect time, exactly which remote functions exist and what arguments each one takes, and validates the call locally before it goes on the wire. A wrong argument type fails in the collector, not as an opaque error from a remote JVM.

Rows are streamed, never buffered

The response is a state machine: Columns, then Row*, then Abort or Error. On the server side that is the Request/Promise/Committer model:

1
2
3
4
5
6
7
pub async fn new_commit(&mut self, columns: Columns) -> Result<Committer<'_>> {
    self.tx.send(State::from(columns.clone())).await?;
    Ok(Committer { tx: &mut self.tx, args: &self.args, columns })
}
pub async fn send(&mut self, state: State) -> Result<()> { self.tx.send(state).await?; Ok(()) }
pub async fn error(&mut self, err: Error) -> Result<()> { self.tx.send(State::from(err)).await?; Ok(()) }
pub async fn ok(&mut self) -> Result<()> { self.tx.send(State::Ok).await?; Ok(()) }

new_commit announces the column names and hands back a Committer, the object that then sends each row, an error, or a final “ok”. tx is a channel: the producer (the query) puts states in, the consumer (the writer) takes them out.

libcore/rust/src/request.rs:63. The producer and the consumer are connected by an mpsc channel, so a query that returns ten million rows never materialises in memory on either side, the timeout applies per row, and the client can abort mid-stream.

Calling a service from Lua

The scripting environment is where the microservices become usable. A script declares which schema it needs, and gets a handle whose methods are exactly the remote service’s advertised datasources:

1
2
3
4
5
6
7
local schema_proxy = { proxy = schema, instance = schema:get_instance() };
for _, call_name in pairs(schema_proxy.proxy:get_datasources()) do
    schema_proxy[call_name] = function(...)
        return schema_proxy.proxy:call(call_name, request:get_uid(), ...);
    end
end
_G[name] = schema_proxy;

The proxy is built at run time from the signature the server advertised at connect time: for every datasource the server listed, the script gets a method of the same name. The original loop defines two methods per datasource — the one shown, and a try_-prefixed twin that returns {code, msg, statement} instead of raising a Lua error when the call fails. request:get_uid() is the id of the current request, which the service uses to send the answer back to the right waiter. _G[name] = schema_proxy publishes the handle as a global, so oracle:query(sql, 20) becomes a statement request.

A script that needs to work whether the host was reached by SSH or by a native agent asks for both, and takes whichever is bound:

1
2
3
4
5
6
7
local function select_host(func)
    local statement = nil;
    match_schema("dg",    function () statement = func(dg) end)
    match_schema("host",  function () statement = func(host) end)
    match_schema("agent", function () statement = func(agent) end)
    return statement;
end

match_schema(schema, callback) asks the session whether a schema is available (session:is_find_schema); if it is, it builds the handle (schemas(schema)) and runs the callback. select_host then tries the three host-like schemas in turn — Data Guard, SSH host, native agent — and keeps the first statement that comes back. The caller’s code is identical on all three paths; only the transport behind the handle differs.

The timeout is the last positional argument of every remote call, and the argument marshalling is explicit:

1
2
3
4
5
6
7
8
fn from_lua_multi(values: MultiValue<'lua>, lua: &'lua Lua) -> mlua::Result<Self> {
    let mut args = Args::from_lua_multi(values, lua)?;
    let timeout: u32 = args.remove(args.len() - 1)?;   // timeout = last arg
    let name: String = args.remove(0)?;                // datasource name
    let uid: String = args.remove(0)?;                 // request uid
    args.remove_index(0); // <-- leftover, see §3
    Ok(CallArgs { name, uid, args, timeout })
}

This is the body of the FromLuaMulti implementation for CallArgs — the trait that converts a list of Lua values into one Rust value. It peels arguments off the front and the back: the timeout is last, the datasource name and the request uid are first. The last line is a leftover from before the request-uid slot existed, and it shows the fragility of an implicit positional convention across a language boundary: the client sees [name, uid, ...args, timeout], so remove_index(0) throws away the first real argument — and panics outright for a zero-argument datasource.

Where the boundaries actually are

The boundary is not “metrics versus commands”. It is who owns the transport:

  • The agent serves filesystem, cpu_usage, memory_usage, swap_usage, iops_usage, diskus, io_counters, host_basic, os_info, check_dir, read_file, write_file, mkdir, shell, ping, watch_log/unwatch_log.
  • The host schema serves only read_file, upload, shell, ping — because that is all SSH can give you. It has no filesystem and no diskus, which is why host.lua re-implements df -Pk parsing in Lua while the agent path calls a native agent:filesystem(10).
  • The database schema serves query, update, test, clean_sql_conditions — one generic SQL verb, and all the vendor semantics (AWR, v$session, InnoDB metrics) live in the Lua tasks on the collector side. The service only knows how to run SQL and hand back rows.

So the interesting logic is in the scripts, and the services are mechanism. That split is what makes it possible to add a metric for a database that has a driver but no new service code, and to add a whole new transport without touching a single task.

Failure handling

Service down. SchemaClient::connect fails with a typed error. The instance stays registered, session:find_schema returns connection_refused, match_schema simply does not fire the callback, and session:schema_error() reports the code. A monitor timer checks the connection state every 40 seconds, re-dials, and publishes the instance’s state to Riemann.

Timeout. The client sends timeout - 1 on the wire, so the callee fails one second before the caller’s own deadline. That small trick avoids the classic deadlock where both sides time out simultaneously and nobody knows why. The server enforces it, the agent wraps cmd.output() in a timeout, and the Java ShellDataSource polls the SSH channel until the deadline and throws.

Errors are data. A wire error carries a typed code — the code is base | (index << 8), with ten bases and about sixty named codes. Lua translates them into ten business codes, and the health-check helper turns a failure into a metric row: request:commit({ metric = handler_error_code(resp.code), message = resp.msg }).

A monitoring system needs its own failures to be monitorable, so a failed collection is a data point, not an exception that vanishes.

A broken pipe fans out. The reader task marks the client closed and pushes the error into every pending request channel, so no caller hangs waiting for a reply that can never arrive.

Trade-offs

Process isolation buys dependency isolation and costs a network hop. The agent needs heim/diskus; the database service needs four JDBC drivers; neither pollutes the other, and a JVM crash in JDBC code cannot take the collector down. The price is a round trip and a second runtime per hop.

One protocol, many languages is a permanent drift risk. The frame format, the state machine and the error codes are implemented twice — once in Rust, once in Java — and documented in doc/src/protocol/ rather than only in code. The benefit is that a new service can be written in any language with a ~500-line codec, and the collector never learns what is on the other end.

The same binary must speak both roles. bee-codec has client and server features and the collector uses both. The agent does too — it serves the agent schema and connects back to the collector’s bee schema to pull log-watch targets. One implementation of framing, one of error mapping, one of the request-id demultiplexer — and one place where a bug hits both directions.

The client is not the server, and the client is the one with a bug. The remove_index(0) leftover above is the kind of defect that only appears when the two halves of a positional convention drift apart.

What I would do differently

Name the argument convention, or use a table. oracle:query(sql, 20) with a trailing timeout is compact and completely implicit. A named form (query{ sql = ..., timeout = 20 }) would have made the leftover line impossible to write and the failure mode obvious.

Put the argument validation where the signature is. The connect handshake already advertises name → Vec<DataType>; the client should reject a bad call before it becomes a wire request, with a message that names the argument.

Let the service own the semantics it is uniquely able to own. The database service already knows how to speak JDBC; letting it also own the “give me the current sessions for this instance” query would move work off the collector and shorten the round trip for the metrics that matter most. The rule “policy in the script, mechanism in the service” is right, but the line between those two is worth drawing per datasource rather than globally.

Version the protocol in the handshake. ConnectResp::Ok carries a version field, and the client checks it — but the frames themselves are not versioned, so an old service and a new collector can disagree about a payload and only find out at runtime.

And keep the byte-stream escape hatch. It is the reason the protocol never needed a schema registry, and it is the reason adding a new kind of service — a log shipper, a metrics bridge, a GPU host — is a 500-line job rather than a negotiation.

References

  • tokio — the Rust runtime the collector runs on
  • sshj — the Java SSH library behind the host service
  • heim and diskus — the crates the native agent reads the machine with
  • mlua — the Rust binding that runs the Lua tasks
  • riemann — the time-series database the metrics go to
  • bytes — the byte-buffer crate behind the Bytes escape hatch
  • SSH Transport Layer (RFC 4253) — the protocol the host service speaks