How to Use Ktor's Pipeline in a Real Service
A monitoring collector needs two things that look like one thing: an HTTP API for operators, and a pipeline of work that runs when an API call asks for a metric. In this project they are genuinely two pipelines, in two modules, that never share a single object — and getting that separation right was most of the design.
The interesting part is that the core copied Ktor’s pipeline implementation to stay framework-free, and then the two copies drifted apart in instructive ways.
What this post is about. It is about Ktor’s pipeline — an ordered list of stages; each stage holds a few functions that run in turn; an interceptor is one of those functions — and about what happens when a project needs the same idea in two places. You will learn how Ktor assembles an HTTP server out of installed plugins, how the collector’s own copy of the pipeline works, and where the two leak into each other.
What you need to know first. Ktor is a Kotlin web framework that runs on Netty, the Java networking library, and its handlers are coroutines: functions that can pause and resume instead of blocking a thread. Nothing else here is special-cased.
The situation
The product is an always-on collector that pulls metrics from customer databases and hosts. It has to expose an HTTP control plane: register instances, run a task on demand, inspect a connection, watch the logs, check health. That API is a thin shell over a core that does the real work.
The constraint that shaped everything: the core must not depend on Ktor. Its build.gradle declares dependencies { } — empty. The core has to be embeddable in any host application, including ones that are not HTTP services at all, and the same core has to run under a different vendor’s commercial product.
So the HTTP layer lives in the app module, and the collection layer lives in the core, and neither imports the other.
Why Ktor for the HTTP side
The alternatives were Spring Boot, plain Netty, or Ktor.
Spring Boot would have brought a DI container, classpath scanning, Tomcat and an actuator — for a service whose whole point is to be a single shaded jar dropped onto a customer host. Ktor gives coroutine-native handlers, an embedded Netty server with no servlet container, and a plugin model that composes with the way the author already thought about pipelines.
Plain Netty would have meant hand-writing HTTP parsing, routing, content negotiation and error mapping — which is exactly the ~140 lines that HttpModule.kt buys.
The application is assembled programmatically rather than through EngineMain (Ktor’s own entry point that reads configuration from Ktor’s hierarchy), because the configuration does not come from that hierarchy at all: a custom HOCON file is read by Typesafe Config, rendered to JSON and parsed by Gson into a JsonObject, then digested into typed config classes. Using EngineMain would have meant two competing sources of truth. The server is started with embeddedServer(Netty, ...), and the Netty group sizes are hand-tuned: half the cores for connections and workers, all of them for calls, because a call is the expensive part.
The plugin stack
Ktor’s pipeline is a sequence of phases, each holding interceptors. This application installs six features, in this order:
1 | fun Application.module(register: ScrapeRegister, config: ApplicationConfig) { |
install adds a feature: a bundle of interceptors that Ktor merges into its pipeline at the right place. DefaultHeaders adds Date/Server to every response. DropwizardMetrics collects Ktor’s own ktor.calls.* timers. StatusPages turns exceptions into responses. ContentNegotiation handles JSON. Authentication handles credentials. Routing holds the five route groups. The { ... } blocks are the per-feature configuration; the StatusPages and Routing ones are filled in further down.
Not installed, and worth noting: no CallLogging, no CORS, no Compression, no CallId, no Sessions, no RequestValidation.
StatusPages is where the error contract lives
1 | exception<ScrapeException> { cause -> |
exception<T> registers a handler for one exception type: when a request throws it, the handler builds the response. call.respond sends that response; mapOf builds a two-entry map that is serialised as a JSON object, so every error body has a numeric code and a message. The first handler is for the application’s own ScrapeException; the second catches everything else.
Ktor’s handler lookup is exact class first, then superclasses, then interfaces, so ScrapeCollectException (a subclass of ScrapeException) lands on the right handler regardless of registration order.
The catch-all is the interesting part: every unexpected error becomes a 400 Bad Request. A database that fell over and a client that sent a malformed body produce the same status code. The body does carry a code (500001 for “other”), but every client that only looks at the status code cannot tell them apart.
ContentNegotiation decides what a missing field costs you
ContentNegotiation is installed with Jackson, not kotlinx-serialization, and jackson { enable(SerializationFeature.INDENT_OUTPUT) } turns on pretty-printed JSON. jacksonObjectMapper() brings the Kotlin module, which means FAIL_ON_UNKNOWN_PROPERTIES stays on — so a request body missing a non-default constructor parameter throws MissingKotlinParameterException, which the catch-all turns into a 400. This is why the route tests assert BadRequest for some inputs: the DTOs are strict, and strictness is expressed as a client error.
Authentication is installed and permanently skipped
basic { skipWhen { true } } disables the basic-auth provider outright. In Ktor 1.3.0, skipWhen { true } takes the “fast path” branch: the provider is never merged into the pipeline, no challenge is issued, and the validate block — the code that would compare the password with config.httpConfig.authConfig.password and return a UserIdPrincipal on a match — is dead code. AuthConfig.enabled is parsed from the config file and then never consulted anywhere. The API is effectively unauthenticated, and the code reads as if it were not.
A second pipeline, in the core
Since the core cannot depend on Ktor, it carries its own copy. core/common/Pipelines.kt is a near-verbatim port of Ktor 1.3.0’s io.ktor.util.pipeline.Pipeline — PhaseContent, PipelinePhaseRelation (Before/After/Last), findPhase, cacheInterceptors, tryAddToPhaseFastpath, fastPathMerge, merge. Only the header differs (Ktor’s Apache notice was replaced by the company’s). core/common/PipelineContext.kt ports SuspendFunctionGun, the delicate interceptor runner: it starts each interceptor as a coroutine, returns control if the interceptor suspends (for example while it waits for a database), and resumes the pipeline with the failure if the interceptor throws.
There are three domain pipelines, and their phases are named for the domain, not for HTTP:
1 | class ScrapeCollectPipeline : Pipeline<ScrapeResponseData, ScrapeCollectCall>( |
A PipelinePhase is a named slot that holds a list of interceptors, and the constructor takes the phases in the order the work should happen: check the request, transform it, render it, validate it, collect it, mark it collected, and clean up. The two type parameters are the response type and the call type the pipeline works with. There is also a request pipeline (Before → Transform → After → Send) and a connect pipeline (Connect → Connected).
The port also drops two things: upstream’s recoverStackTraceBridge and resumeWithException, so exceptions crossing a pipeline phase lose stack-trace recovery.
Each phase is filled by a different feature: the engine installs the request pipeline’s Transform (the fan-out over tasks × instances), the collector installs all seven collect phases, the connection manager installs the two connect phases. A plugin can only add an interceptor to an existing phase, keyed by task id — the merge() machinery that would let it insert a whole new phase exists in the ported code and is used by nothing except the ported test.
Two pipelines, one seam
1 | ┌─ HTTP ───────┐ ┌──── core ────────────────────┐ |
The left box is Ktor’s own pipeline, the right box is the core’s three pipelines — requestPipeline, collectPipeline and connectPipeline, the last one running Connect → Connected. The arrow pointing right is the request travelling from the HTTP layer into the core; the arrow pointing left is the answer travelling back, as a DTO (a data-transfer object: a plain class whose fields become the JSON body). The lines of text inside the core box are the phases of each pipeline, in the order they run.
The two pipelines never share a PipelineContext or an Attributes container. The seam is three ordinary function calls: route → register.collect { … } → engine.requestTasks → requestPipeline.execute → collectPipeline.execute → connectPipeline.execute → a ScrapeResponse → a DTO → call.respond.
That separation is a feature: HTTP concerns stay in the app, collection concerns stay in the core, and neither drags the other along. The cost is that cross-cutting concerns get implemented twice, and Ktor’s metrics never see the domain work.
Where the blocking work leaks
Ktor 1.3.0’s Netty engine runs each call on the channel’s event loop. The heavy JDBC and SSH work is therefore pushed onto dedicated pools — “Scrape” for collection, “Instance-Pool” for connections, “Monitor” for the scheduler, “Riemann” for shipping. The load-bearing line is one operator:
1 | register.requestPipeline.intercept(ScrapeRequestPipeline.Transform) { |
intercept adds an interceptor to a phase; proceedWith hands the result to the next interceptor. This is the body of the engine’s install function: it adds one interceptor to the Transform phase of the request pipeline. withContext runs its block on a different coroutine context — a context is the bundle of “which thread pool, which logger, which deadline” that a coroutine carries around. The + operator merges two contexts, and CoroutineContext.plus lets the right operand win, so the pipeline’s own context — which is the root continuation’s context, i.e. the Ktor call’s context, with the Netty dispatcher — overrides the engine’s pool.
It works only because the blocking work sits inside async { } blocks that carry the engine’s dispatcher, and the event loop merely suspends on awaitAll(). Swap the two operands and the intent holds.
Two routes block the event loop outright: GET /instance/publicKey runs Runtime.getRuntime().exec(...).waitFor(), and POST /instance/test calls driver.createConnection(...).checkLive() straight from the handler, bypassing the pipeline entirely.
Trade-offs
The duplicated pipeline is ~490 lines of forked infrastructure that has to be tracked across Kotlin and Ktor upgrades, and it re-implements semantics (suspendCoroutineUninterceptedOrReturn) that are easy to get subtly wrong. The Apache-2.0 notice on the original was replaced rather than preserved, which is the kind of detail that a company’s legal team notices and an engineer does not.
The catch-all 400 makes server faults look like client faults. Every unhandled exception — including a JVM out-of-memory in a collector — is a 400.
DropwizardMetrics collects Ktor’s timers and then throws them away. The reporter is created with the wrong registry:
1 | fun createReport(registry: MetricRegistry): ScheduledReporter = |
registry is the registry Ktor’s own timers go into, and metrics is the domain’s own registry; the reporter publishes the second one, so the ktor.calls.* metrics are recorded and never published. Only the domain’s registry is reported.
The strict Jackson mapper turns “field missing” into “request rejected”, which is a reasonable choice but couples the wire format to the Kotlin constructor signature: adding a field to a DTO with a default is safe, changing one without a default breaks every older client.
What I would do differently
Decide once whether the core is framework-free, and pay for it honestly. The right answer here was “yes, framework-free” — but the cost should have been a small pipeline of my own with my own names, not a byte-for-byte port of Ktor’s class with a different header. A 60-line phase/interceptor implementation would have covered everything this project uses, and the phases could have been designed for the domain (Render, Valid) instead of reproducing Ktor’s internals.
Use the Ktor pipeline properly. StatusPages handles the domain exceptions; ContentNegotiation handles DTOs; the routes are thin. What is missing is CallLogging (there is an application.log.info inside the route instead), RequestValidation (the DTO constructor is the validation, which is why errors surface as 400s), and Compression.
Fix the dispatcher precedence and move the blocking routes off the event loop. withContext with the engine context on the left would express the intent correctly, and GET /instance/publicKey and POST /instance/test belong on a bounded pool.
Give the metrics a real registry. createReport should report the registry Ktor actually populates, or the whole DropwizardMetrics feature should go — a feature that silently discards its data is worse than no feature.
And delete the dead authentication. Either honour AuthConfig.enabled and make skipWhen conditional, or remove the Authentication installation. A half-installed auth provider is a comment that lies.
The part I would keep exactly as it is: the layering. HTTP concerns in the app module, collection concerns in the core, and a core that compiles against an empty dependencies { } block. That constraint is what makes the collector embeddable, and it is worth more than the 490 lines it cost.
References
- ktor — the Kotlin framework whose pipeline this post is about
- Ktor documentation — the entry point for everything below
- Plugins — how
installassembles features into the pipeline - Routing — how the five route groups are declared
- Status pages — how
exception<T>handlers become responses - Serialization —
ContentNegotiationand the Jackson configuration - Engines —
embeddedServer(Netty, …)and the group sizes - Kotlin coroutines — the suspension model the whole pipeline rests on