Continue conversation on use of buftok in input plugin

[donoghuc]

Bug

The HTTP input clones the codec per TCP connection, so different connections are isolated. Requests that arrive on the same connection are dispatched to a shared worker pool, so two requests on one connection can run at the same time against that connection's single json_lines BufferedTokenizer. That concurrent access drops lines, duplicates lines, and can throw NoSuchElementException, which the input turns into an HTTP 500. A connection that sends one request at a time is safe. The trigger is concurrent, pipelined requests on one connection.

Environment

Official image docker.elastic.co/logstash/logstash:9.3.4. The plugin under test is logstash-input-http 4.1.7 and the codec is json_lines 3.2.2, both the stock gems inside the image. Pipeline:

input  { http { port => 8080 additional_codecs => { "application/json" => "json_lines" } threads => 8 } }
output { file { path => "/out/events.ndjson" codec => json_lines flush_interval => 1 } }

How the driver builds a request

Each line is a small JSON object whose id is globally unique across the whole run. A request body is a block of complete, newline terminated lines, so the server does exactly one append per request. The framing is a normal HTTP POST with Content-Length and Connection: keep-alive.

def build_request(base, nlines):
    lines = [f'{{"id":{base+i},"pad":"{_pad}"}}' for i in range(nlines)]
    body = ("\n".join(lines) + "\n").encode()
    head = (f"POST / HTTP/1.1\r\nHost: {HOST}\r\n"
            f"Content-Type: application/json\r\n"
            f"Content-Length: {len(body)}\r\n"
            f"Connection: keep-alive\r\n\r\n").encode()
    return head + body

How the driver forces the concurrency

Everything goes over a single socket, which is a single TCP connection, so every request shares that connection's one cloned codec. The driver keeps a sliding window of WINDOW requests in flight. It first writes WINDOW requests back to back without reading any response, then for every response it reads it sends one more. Because the requests are on the wire before their responses are read, the server's worker pool picks several of them up at once and runs them concurrently against the one buffer.

sock = socket.create_connection((HOST, PORT))   # ONE connection for the whole run

# prime the pipe: put WINDOW requests on the wire without waiting for responses
while sent < N_REQ and in_flight < WINDOW:
    sock.sendall(build_request(sent * LINES_PER_REQ, LINES_PER_REQ))
    sent += 1
    in_flight += 1

# steady state: for each response read, send one more, keeping WINDOW in flight
while acked < sent or sent < N_REQ:
    if in_flight > 0:
        code = read_one_response()            # parses status line + body by Content-Length
        if code == 200:
            accepted_bases.append(resp_idx * LINES_PER_REQ)
        resp_idx += 1
        acked += 1
        in_flight -= 1
    if sent < N_REQ and in_flight < WINDOW:
        sock.sendall(build_request(sent * LINES_PER_REQ, LINES_PER_REQ))
        sent += 1
        in_flight += 1

Only requests that returned HTTP 200 are treated as accepted. The driver records the exact ids in each accepted request, so the accepted set is precise.

with open(ACCEPTED_IDS_FILE, "w") as f:
    for b in accepted_bases:
        for i in range(LINES_PER_REQ):
            f.write(f"{b+i}\n")

The default run is 4000 requests of 200 lines each, with a window of 16. The window is kept under the input's max_pending_requests of 200 so nothing is rejected for backpressure.

How loss is measured

The output file holds one JSON line per emitted event. The analyzer compares the set of accepted ids against the set of emitted ids. Anything accepted over HTTP but missing from the output is silent loss. A repeated id is a duplicate.

accepted = set(int(l) for l in open(accepted_ids_file))

emitted = set()
for line in open(output_ndjson):
    emitted.add(int(json.loads(line)["id"]))

missing = accepted - emitted     # accepted (HTTP 200) but never emitted -> silent loss

The run also reads the pipeline events.in counter from the monitoring API. Because a tokenizer drop happens before an event is created, events.in falling short of the accepted line count places the loss inside the codec rather than in the output.

The fix

The change is in decode_body in lib/logstash/inputs/http.rb. Give each request its own codec so its buffer is private and cannot be shared with a concurrent request.

-    codec = additional_codecs.fetch(HttpUtil.getMimeType(content_type), default_codec)
+    codec = additional_codecs.fetch(HttpUtil.getMimeType(content_type), default_codec).clone

This works because codec.clone re-instantiates the plugin and builds a fresh BufferedTokenizer, so nothing stateful is shared across concurrent requests.

How each build was tested

Three builds were driven through the identical single connection pipelined load and the same analyzer, so the only variable between rows is the build itself. Each build is a fresh container running the pipeline above, then the same driver, then the same accepted-versus-emitted comparison.

No fix. The stock image, unmodified. The harness starts the container and runs the load with nothing patched. This is the bug as it ships.

Logstash-core side fix. This approach adds synchronized to the tokenizer's append and flush methods. The other public methods are already synchronized. It was tested in isolation by taking the exact BufferedTokenizer.java from the image, adding only those two keywords, compiling it against the image's own log4j jar, and swapping the resulting class files into a copy of logstash-core.jar that is mounted over the container's. Because the two keywords are the only difference from stock, this isolates the effect of that change. A recompiled but otherwise unmodified build reproduced the stock loss exactly, which confirms the swap itself changes nothing else.

Plugin side fix. The per-request clone in decode_body shown above. It was tested by mounting the patched http.rb over the plugin gem file in the container, with everything else identical.

Results

Single connection pipelined load, 4000 requests of 200 lines each, repeated runs.

build                            accepted     silent loss      duplicates   HTTP 500
no fix (stock)                   796,200      243,352 (31%)    67,437       15
logstash-core side fix (sync)    ~798,000     ~2 to 6%         0            8 to 11
plugin side fix (clone per req)  800,000      0                0            0

The logstash-core side fix removes the memory corruption, so duplicates and garbled tokens go to zero, but it does not stop the silent loss or the NoSuchElementException path. Locking each method does not make the multi call decode then flush sequence atomic, and the shared buffer remains. The plugin side fix removes all of it. With the plugin fix every one of the 4000 requests returned HTTP 200, the pipeline events.in delta equaled the 800,000 accepted lines, and the output contained all 800,000 ids with no duplicates.

appendix

the repro code and attempt at profiling is at gist.github.com/donoghuc/a047177d3461ffce414ef908c445960d

[github-actions]

Changelog and Version Management

No changelog or version changes detected. You can either update them manually or use a comment command to update them automatically on merge:

/pr_prepare release:<patch|minor|major> — bump version + update changelog
/pr_prepare changelog:add — changelog entry only (no version bump)

/pr_prepare release:patch
- Add the changelog entry here
  - (Additional details can be provided in multiple lines, respecting markdown formatting)

Omit the entry line to use the PR title. If multiple commands are posted, the last one wins.