From 90cdee06413fdd60a576e0906bb4e56582561cb3 Mon Sep 17 00:00:00 2001 From: "wrossmorrow-pi-bot[bot]" <346594324+wrossmorrow-pi-bot[bot]@users.noreply.github.com> Date: Tue, 27 Oct 2026 23:26:42 +0000 Subject: [PATCH 01/10] feat: add plain gRPC load generator for ext proc exchanges --- .../wrossmorrow/extproc/loadgen/LoadGen.java | 311 ++++++++++++++++++ 1 file changed, 311 insertions(+) create mode 100644 src/main/java/com/github/wrossmorrow/extproc/loadgen/LoadGen.java diff --git a/src/main/java/com/github/wrossmorrow/extproc/loadgen/LoadGen.java b/src/main/java/com/github/wrossmorrow/extproc/loadgen/LoadGen.java new file mode 100644 index 0000000..183eb45 --- /dev/null +++ b/src/main/java/com/github/wrossmorrow/extproc/loadgen/LoadGen.java @@ -0,0 +1,311 @@ +package com.github.wrossmorrow.extproc.loadgen; + +import build.buf.gen.envoy.config.core.v3.HeaderMap; +import build.buf.gen.envoy.config.core.v3.HeaderValue; +import build.buf.gen.envoy.service.ext_proc.v3.ExternalProcessorGrpc; +import build.buf.gen.envoy.service.ext_proc.v3.HttpBody; +import build.buf.gen.envoy.service.ext_proc.v3.HttpHeaders; +import build.buf.gen.envoy.service.ext_proc.v3.ProcessingRequest; +import build.buf.gen.envoy.service.ext_proc.v3.ProcessingResponse; +import com.google.protobuf.ByteString; +import io.grpc.ManagedChannel; +import io.grpc.ManagedChannelBuilder; +import io.grpc.stub.StreamObserver; +import java.time.Duration; +import java.time.Instant; +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; +import java.util.logging.Logger; + +/** + * Plain gRPC load generator for the envoy ext_proc SDK (Java). + * + *

Drives full ext_proc exchanges over a raw bidirectional gRPC stream to the SDK's ext_proc + * server (`envoy.service.ext_proc.v3.ExternalProcessor/Process`), without an envoy in the path, and + * reports per-exchange latency and errors. + * + *

Each exchange mirrors the phase sequence the envoy ext_proc filter drives for a buffered + * request/response: request headers -> request body -> response headers -> response body, then the + * stream is closed. Because envoy (and the HTTP machinery in between) are absent, the reported + * latency is the processor-side cost of the SDK's gRPC server path, complementing the containerized + * k6 load tests in `loadtest` that drive the full envoy -> extproc -> upstream chain. + */ +public class LoadGen { + + private static final Logger logger = Logger.getLogger(LoadGen.class.getName()); + + /** + * The response header the server may set (via the ProcessingOptions.UpdateDurationHeader option) + * carrying the accumulated extproc processing time in nanoseconds. + */ + static final String DURATION_HEADER = "x-extproc-duration-ns"; + + static final String DEFAULT_ADDR = "127.0.0.1:50051"; + static final int DEFAULT_ITERATIONS = 100; + static final int DEFAULT_CONCURRENCY = 10; + + public static void main(String[] args) throws Exception { + String addr = DEFAULT_ADDR; + int iterations = DEFAULT_ITERATIONS; + int concurrency = DEFAULT_CONCURRENCY; + for (int i = 0; i < args.length; i++) { + String arg = args[i]; + if (arg.startsWith("--addr=")) { + addr = arg.substring("--addr=".length()); + } else if (arg.startsWith("--iterations=")) { + iterations = Integer.parseInt(arg.substring("--iterations=".length())); + } else if (arg.startsWith("--concurrency=")) { + concurrency = Integer.parseInt(arg.substring("--concurrency=".length())); + } + } + + if (iterations <= 0) { + System.err.println("LoadGen: iterations must be greater than zero"); + System.exit(1); + } + + System.out.println("LoadGen: target extproc gRPC server: http://" + addr); + System.out.println( + "LoadGen: running " + iterations + " exchanges with concurrency " + concurrency); + + int port = 50051; + String host = addr; + int colon = addr.lastIndexOf(':'); + if (colon >= 0) { + host = addr.substring(0, colon); + port = Integer.parseInt(addr.substring(colon + 1)); + } + ManagedChannel channel = + ManagedChannelBuilder.forAddress(host, port).useTransportSecurity().build(); + ExternalProcessorGrpc.ExternalProcessorStub stub = ExternalProcessorGrpc.newStub(channel); + ExecutorService executor = Executors.newFixedThreadPool(concurrency); + try { + List workerResults = new ArrayList<>(); + List> futures = new ArrayList<>(); + for (int i = 0; i < concurrency; i++) { + final int base = i; + final ExternalProcessorGrpc.ExternalProcessorStub workerStub = stub; + final int workerIterations = iterations; + final int workerConcurrency = concurrency; + futures.add( + executor.submit( + () -> runWorker(workerStub, base, workerConcurrency, workerIterations))); + } + for (Future future : futures) { + workerResults.add(future.get(30, TimeUnit.SECONDS)); + } + + List latencies = new ArrayList<>(); + List extprocDurationsNs = new ArrayList<>(); + List errors = new ArrayList<>(); + for (WorkerResult result : workerResults) { + latencies.addAll(result.latencies); + extprocDurationsNs.addAll(result.extprocDurationsNs); + errors.addAll(result.errors); + } + latencies.sort((left, right) -> left.compareTo(right)); + double avgMs = 0.0; + if (!latencies.isEmpty()) { + Duration sum = Duration.ZERO; + for (Duration d : latencies) { + sum = sum.plus(d); + } + avgMs = sum.toMillis() / (double) latencies.size(); + } + Double p50Ms = percentile(latencies, 0.50); + Double p95Ms = percentile(latencies, 0.95); + double maxMs = latencies.isEmpty() ? 0.0 : latencies.get(latencies.size() - 1).toMillis(); + double avgDurationMs = 0.0; + if (!extprocDurationsNs.isEmpty()) { + long sumNs = 0; + for (long ns : extprocDurationsNs) { + sumNs += ns; + } + avgDurationMs = sumNs / ((double) extprocDurationsNs.size() * 1e6); + } + + for (ExchangeError e : errors) { + System.out.println("LoadGen: exchange " + e.index + " failed: " + e.error); + } + System.out.println( + "LoadGen: finished: " + latencies.size() + " exchanges, " + errors.size() + " errors"); + System.out.println( + "LoadGen: latency avg=" + + avgMs + + "ms p50=" + + p50Ms + + "ms p95=" + + p95Ms + + "ms max=" + + maxMs + + "ms"); + System.out.println( + "LoadGen: extproc duration avg=" + + avgDurationMs + + "ms (" + + extprocDurationsNs.size() + + " samples)"); + + if (!errors.isEmpty()) { + System.exit(1); + } + } finally { + executor.shutdownNow(); + } + + channel.shutdownNow(); + } + + static WorkerResult runWorker( + ExternalProcessorGrpc.ExternalProcessorStub stub, int base, int stride, int iterations) { + WorkerResult result = new WorkerResult(new ArrayList<>(), new ArrayList<>(), new ArrayList<>()); + int index = base; + while (index < iterations) { + ExchangeOutcome outcome = runExchange(stub); + if (outcome.error != null) { + result.errors.add(new ExchangeError(index, outcome.error)); + } else { + result.latencies.add(outcome.latency); + result.extprocDurationsNs.add(outcome.extprocDurationNs); + } + index += stride; + } + return result; + } + + static ExchangeOutcome runExchange(ExternalProcessorGrpc.ExternalProcessorStub stub) { + Instant start = Instant.now(); + ResponseCollector responseObserver = new ResponseCollector(); + StreamObserver requestObserver = stub.process(responseObserver); + + requestObserver.onNext( + ProcessingRequest.newBuilder() + .setRequestHeaders( + HttpHeaders.newBuilder() + .setHeaders(requestHeaderMap("POST", "/test")) + .setEndOfStream(false) + .build()) + .build()); + requestObserver.onNext( + ProcessingRequest.newBuilder() + .setRequestBody( + HttpBody.newBuilder() + .setBody(ByteString.copyFromUtf8("{}")) + .setEndOfStream(true) + .build()) + .build()); + requestObserver.onNext( + ProcessingRequest.newBuilder() + .setResponseHeaders(HttpHeaders.newBuilder().setEndOfStream(true).build()) + .build()); + requestObserver.onNext( + ProcessingRequest.newBuilder() + .setResponseBody( + HttpBody.newBuilder() + .setBody(ByteString.copyFromUtf8("{}")) + .setEndOfStream(true) + .build()) + .build()); + requestObserver.onCompleted(); + + Long extprocDurationNs = extractDurationNs(responseObserver); + Duration latency = Duration.between(start, Instant.now()); + String error = responseObserver.error == null ? null : responseObserver.error.getMessage(); + return new ExchangeOutcome(latency, extprocDurationNs, error); + } + + static Long extractDurationNs(ResponseCollector responseObserver) { + if (responseObserver.lastResponse == null) { + return null; + } + if (responseObserver.lastResponse.hasOverrideMessageTimeout()) { + long seconds = responseObserver.lastResponse.getOverrideMessageTimeout().getSeconds(); + int nanos = responseObserver.lastResponse.getOverrideMessageTimeout().getNanos(); + return Long.valueOf(seconds * 1000000000L + nanos); + } + return null; + } + + static HeaderMap requestHeaderMap(String method, String path) { + HeaderMap.Builder builder = HeaderMap.newBuilder(); + builder.addHeaders( + HeaderValue.newBuilder() + .setKey(":method") + .setRawValue(ByteString.copyFromUtf8(method)) + .build()); + builder.addHeaders( + HeaderValue.newBuilder() + .setKey(":path") + .setRawValue(ByteString.copyFromUtf8(path)) + .build()); + return builder.build(); + } + + static double percentile(List sorted, double percentile) { + if (sorted.isEmpty()) { + return 0.0; + } + int rank = (int) Math.round((double) (sorted.size() - 1) * percentile); + return sorted.get(rank).toMillis(); + } + + static final class ResponseCollector implements StreamObserver { + ProcessingResponse lastResponse; + Throwable error; + boolean completed; + + @Override + public void onNext(ProcessingResponse value) { + lastResponse = value; + } + + @Override + public void onError(Throwable t) { + error = t; + } + + @Override + public void onCompleted() { + completed = true; + } + } + + static final class WorkerResult { + final List latencies; + final List extprocDurationsNs; + final List errors; + + WorkerResult( + List latencies, List extprocDurationsNs, List errors) { + this.latencies = latencies; + this.extprocDurationsNs = extprocDurationsNs; + this.errors = errors; + } + } + + static final class ExchangeError { + final int index; + final String error; + + ExchangeError(int index, String error) { + this.index = index; + this.error = error; + } + } + + static final class ExchangeOutcome { + final Duration latency; + final Long extprocDurationNs; + final String error; + + ExchangeOutcome(Duration latency, Long extprocDurationNs, String error) { + this.latency = latency; + this.extprocDurationNs = extprocDurationNs; + this.error = error; + } + } +} From 8eae2fe4e56b9a6198f11fe0affaf3c3b332df8b Mon Sep 17 00:00:00 2001 From: "wrossmorrow-pi-bot[bot]" <346594324+wrossmorrow-pi-bot[bot]@users.noreply.github.com> Date: Tue, 27 Oct 2026 23:30:56 +0000 Subject: [PATCH 02/10] feat: add testing request processor for harness parity --- .../processors/TestingRequestProcessor.java | 270 ++++++++++++++++++ 1 file changed, 270 insertions(+) create mode 100644 src/main/java/com/github/wrossmorrow/extproc/processors/TestingRequestProcessor.java diff --git a/src/main/java/com/github/wrossmorrow/extproc/processors/TestingRequestProcessor.java b/src/main/java/com/github/wrossmorrow/extproc/processors/TestingRequestProcessor.java new file mode 100644 index 0000000..eafaf62 --- /dev/null +++ b/src/main/java/com/github/wrossmorrow/extproc/processors/TestingRequestProcessor.java @@ -0,0 +1,270 @@ +package com.github.wrossmorrow.extproc.processors; + +import build.buf.gen.envoy.service.ext_proc.v3.ProcessingRequest.RequestCase; +import com.fasterxml.jackson.annotation.JsonProperty; +import com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.ObjectMapper; +import com.github.wrossmorrow.extproc.ProcessingOptions; +import com.github.wrossmorrow.extproc.RequestContext; +import com.github.wrossmorrow.extproc.RequestProcessor; +import com.github.wrossmorrow.extproc.RequestProcessorHealthManager; +import java.util.List; +import java.util.Map; +import java.util.logging.Level; +import java.util.logging.Logger; + +/** + * Request processor used by the testing harness. Mirrors the Go and Rust SDKs' tester processors, + * which drive full ext_proc exchanges (via the load generator) and mutate the stream through the + * processing actions encoded in the request body. + */ +public class TestingRequestProcessor implements RequestProcessor { + + private static final Logger logger = Logger.getLogger(TestingRequestProcessor.class.getName()); + + private static final ObjectMapper MAPPER = new ObjectMapper(); + + /** Request processing actions applied by the tester across the phases of a single exchange. */ + public static final class TesterRequestBody { + + @JsonProperty("append_request_headers") + public Map appendRequestHeaders; + + @JsonProperty("add_request_headers") + public Map addRequestHeaders; + + @JsonProperty("overwrite_request_headers") + public Map overwriteRequestHeaders; + + @JsonProperty("remove_request_headers") + public List removeRequestHeaders; + + @JsonProperty("clear_request_body") + public boolean clearRequestBody; + + @JsonProperty("replace_request_body") + public String replaceRequestBody; + + @JsonProperty("append_response_headers") + public Map appendResponseHeaders; + + @JsonProperty("add_response_headers") + public Map addResponseHeaders; + + @JsonProperty("overwrite_response_headers") + public Map overwriteResponseHeaders; + + @JsonProperty("remove_response_headers") + public List removeResponseHeaders; + + @JsonProperty("clear_response_body") + public boolean clearResponseBody; + + @JsonProperty("replace_response_body") + public String replaceResponseBody; + + @JsonProperty("cancel_request") + public boolean cancelRequest; + + @JsonProperty("cancel_request_status") + public int cancelRequestStatus; + + @JsonProperty("cancel_request_headers") + public Map cancelRequestHeaders; + + @JsonProperty("cancel_request_body") + public String cancelRequestBody; + } + + private final ProcessingOptions options; + private volatile TesterRequestBody actions; + + public TestingRequestProcessor() { + this(new ProcessingOptions()); + } + + public TestingRequestProcessor(ProcessingOptions options) { + this.options = options; + } + + @Override + public String getName() { + return "testing"; + } + + @Override + public ProcessingOptions getOptions() { + return options; + } + + @Override + public void setHealthManager(RequestProcessorHealthManager health) {} + + @Override + public void shutdown() { + logger.info(TestingRequestProcessor.class.getSimpleName() + " shutting down"); + } + + @Override + public void close(int gracePeriodSeconds) { + logger.info("Closing tester"); + } + + @Override + public void errorHandler(RequestContext ctx, RequestCase phase, Throwable err) { + logger.log(Level.INFO, "Error in phase {0}: {1}", new Object[] {phase, err}); + } + + @Override + public void processRequestHeaders(RequestContext ctx, Map headers) { + reportHeaderParsing(ctx, headers); + + // optional upstream override, exercising clear-route-cache driven upstream + // changes: rewrite the authority and clear the route cache so envoy + // recomputes the route + String upstream = getHeaderIgnoreCase(headers, "x-route-to-upstream"); + if (upstream != null) { + ctx.overwriteHeader(":authority", upstream); + ctx.clearRouteCache(); + } + + ctx.continueAndReplace(); + } + + @Override + public void processRequestBody(RequestContext ctx, String body) { + // NOTE: assuming buffering, not streaming; if buffering we have to + // accumulate the request body chunks sent until the end of stream + + TesterRequestBody parsed = parseActions(body); + ctx.setValue("actions", parsed); + + if (parsed != null && parsed.cancelRequest) { + ctx.cancelRequest( + parsed.cancelRequestStatus, parsed.cancelRequestHeaders, parsed.cancelRequestBody); + return; + } + + if (parsed != null) { + applyRequestHeaderActions(ctx, parsed); + applyRequestBodyActions(ctx, parsed); + } + + ctx.continueAndReplace(); + } + + @Override + public void processRequestTrailers(RequestContext ctx, Map trailers) { + ctx.continueAndReplace(); + } + + @Override + public void processResponseHeaders(RequestContext ctx, Map headers) { + TesterRequestBody parsed = (TesterRequestBody) ctx.getValue("actions"); + if (parsed != null) { + applyResponseHeaderActions(ctx, parsed); + } + ctx.continueAndReplace(); + } + + @Override + public void processResponseBody(RequestContext ctx, String body) { + TesterRequestBody parsed = (TesterRequestBody) ctx.getValue("actions"); + if (parsed != null) { + applyResponseBodyActions(ctx, parsed); + } + ctx.continueAndReplace(); + } + + @Override + public void processResponseTrailers(RequestContext ctx, Map trailers) { + ctx.continueAndReplace(); + } + + /** Reflect on how the SDK parsed the inbound headers back via the request headers. */ + protected void reportHeaderParsing(RequestContext ctx, Map headers) { + ctx.addHeader("x-parsed-str-count", String.valueOf(ctx.getAllRequestHeaders().size())); + String first = ctx.getAllRequestHeaders().get("x-probe-list"); + ctx.addHeader("x-parsed-list-value", first == null ? "" : first); + List values = ctx.getAllRequestHeaders().values("x-probe-list"); + ctx.addHeader("x-parsed-list-count", String.valueOf(values.size())); + List split = ctx.getAllRequestHeaders().splitList("x-probe-list"); + ctx.addHeader("x-parsed-list-split", String.join("|", split)); + } + + protected TesterRequestBody parseActions(String body) { + if (body == null || body.isEmpty()) { + return null; + } + try { + JsonNode node = MAPPER.readTree(body); + if (node == null || node.isNull() || !node.isObject()) { + return null; + } + TesterRequestBody actions = MAPPER.convertValue(node, TesterRequestBody.class); + return actions; + } catch (Exception e) { + logger.log(Level.WARNING, "Error parsing request body: {0}", e.getMessage()); + return null; + } + } + + protected void applyRequestHeaderActions(RequestContext ctx, TesterRequestBody actions) { + if (actions.appendRequestHeaders != null) { + ctx.appendHeaders(actions.appendRequestHeaders); + } + if (actions.addRequestHeaders != null) { + ctx.addHeaders(actions.addRequestHeaders); + } + if (actions.overwriteRequestHeaders != null) { + ctx.overwriteHeaders(actions.overwriteRequestHeaders); + } + if (actions.removeRequestHeaders != null) { + ctx.removeHeaders(actions.removeRequestHeaders); + } + } + + protected void applyResponseHeaderActions(RequestContext ctx, TesterRequestBody actions) { + if (actions.appendResponseHeaders != null) { + ctx.appendHeaders(actions.appendResponseHeaders); + } + if (actions.addResponseHeaders != null) { + ctx.addHeaders(actions.addResponseHeaders); + } + if (actions.overwriteResponseHeaders != null) { + ctx.overwriteHeaders(actions.overwriteResponseHeaders); + } + if (actions.removeResponseHeaders != null) { + ctx.removeHeaders(actions.removeResponseHeaders); + } + } + + protected void applyRequestBodyActions(RequestContext ctx, TesterRequestBody actions) { + if (actions.clearRequestBody) { + ctx.clearBody(); + } else if (actions.replaceRequestBody != null) { + ctx.replaceBody(actions.replaceRequestBody); + } + } + + protected void applyResponseBodyActions(RequestContext ctx, TesterRequestBody actions) { + if (actions.clearResponseBody) { + ctx.clearBody(); + } else if (actions.replaceResponseBody != null) { + ctx.replaceBody(actions.replaceResponseBody); + } + } + + /** Case-insensitive header lookup over the phase header map. */ + protected String getHeaderIgnoreCase(Map headers, String name) { + if (headers == null) { + return null; + } + for (Map.Entry entry : headers.entrySet()) { + if (entry.getKey() != null && entry.getKey().equalsIgnoreCase(name)) { + return entry.getValue(); + } + } + return null; + } +} From c2f72ed22b70afd5ae43e792e13d6b11975c5be0 Mon Sep 17 00:00:00 2001 From: "wrossmorrow-pi-bot[bot]" <346594324+wrossmorrow-pi-bot[bot]@users.noreply.github.com> Date: Wed, 28 Oct 2026 13:43:45 +0000 Subject: [PATCH 03/10] build: add loadGen gradle task to run the extproc load generator --- build.gradle | 7 +++++++ 1 file changed, 7 insertions(+) diff --git a/build.gradle b/build.gradle index 21d42af..920db3a 100644 --- a/build.gradle +++ b/build.gradle @@ -75,6 +75,13 @@ tasks.named('test') { useJUnitPlatform() } +tasks.register('loadGen', JavaExec) { + group = 'verification' + description = 'Run the plain gRPC load generator against an extproc server.' + mainClass = 'com.github.wrossmorrow.extproc.loadgen.LoadGen' + classpath = sourceSets.main.runtimeClasspath +} + test { useJUnit() useJUnitPlatform() From e3de70b2f6115a0b702c70ecc7397450e57a8f96 Mon Sep 17 00:00:00 2001 From: "wrossmorrow-pi-bot[bot]" <346594324+wrossmorrow-pi-bot[bot]@users.noreply.github.com> Date: Wed, 28 Oct 2026 13:45:42 +0000 Subject: [PATCH 04/10] fix: read extproc duration from response header mutation like upstream --- .../wrossmorrow/extproc/loadgen/LoadGen.java | 38 +++++++++++++++---- 1 file changed, 31 insertions(+), 7 deletions(-) diff --git a/src/main/java/com/github/wrossmorrow/extproc/loadgen/LoadGen.java b/src/main/java/com/github/wrossmorrow/extproc/loadgen/LoadGen.java index 183eb45..04f9a4a 100644 --- a/src/main/java/com/github/wrossmorrow/extproc/loadgen/LoadGen.java +++ b/src/main/java/com/github/wrossmorrow/extproc/loadgen/LoadGen.java @@ -2,6 +2,8 @@ import build.buf.gen.envoy.config.core.v3.HeaderMap; import build.buf.gen.envoy.config.core.v3.HeaderValue; +import build.buf.gen.envoy.config.core.v3.HeaderValueOption; +import build.buf.gen.envoy.service.ext_proc.v3.CommonResponse; import build.buf.gen.envoy.service.ext_proc.v3.ExternalProcessorGrpc; import build.buf.gen.envoy.service.ext_proc.v3.HttpBody; import build.buf.gen.envoy.service.ext_proc.v3.HttpHeaders; @@ -212,20 +214,42 @@ static ExchangeOutcome runExchange(ExternalProcessorGrpc.ExternalProcessorStub s .build()); requestObserver.onCompleted(); - Long extprocDurationNs = extractDurationNs(responseObserver); + Long extprocDurationNs = extractDurationNs(responseObserver.lastResponse); Duration latency = Duration.between(start, Instant.now()); String error = responseObserver.error == null ? null : responseObserver.error.getMessage(); return new ExchangeOutcome(latency, extprocDurationNs, error); } - static Long extractDurationNs(ResponseCollector responseObserver) { - if (responseObserver.lastResponse == null) { + static Long extractDurationNs(ProcessingResponse response) { + if (response == null) { return null; } - if (responseObserver.lastResponse.hasOverrideMessageTimeout()) { - long seconds = responseObserver.lastResponse.getOverrideMessageTimeout().getSeconds(); - int nanos = responseObserver.lastResponse.getOverrideMessageTimeout().getNanos(); - return Long.valueOf(seconds * 1000000000L + nanos); + CommonResponse common; + if (response.getResponseCase() == ProcessingResponse.ResponseCase.RESPONSE_HEADERS) { + common = response.getResponseHeaders().getResponse(); + } else if (response.getResponseCase() == ProcessingResponse.ResponseCase.RESPONSE_BODY) { + common = response.getResponseBody().getResponse(); + } else { + return null; + } + if (common == null || !common.hasHeaderMutation()) { + return null; + } + for (HeaderValueOption option : common.getHeaderMutation().getSetHeadersList()) { + HeaderValue header = option.getHeader(); + if (header == null || !DURATION_HEADER.equals(header.getKey())) { + continue; + } + ByteString raw = header.getRawValue(); + if (raw == null || raw.size() == 0) { + return null; + } + try { + return Long.valueOf( + new String(raw.toByteArray(), java.nio.charset.StandardCharsets.UTF_8).trim()); + } catch (Exception e) { + return null; + } } return null; } From d4f4ce6c7e4028c021b0b1a0282de869d89095d3 Mon Sep 17 00:00:00 2001 From: "wrossmorrow-pi-bot[bot]" <346594324+wrossmorrow-pi-bot[bot]@users.noreply.github.com> Date: Wed, 28 Oct 2026 13:51:01 +0000 Subject: [PATCH 05/10] fix: guard null extproc durations in load generator aggregation --- .../java/com/github/wrossmorrow/extproc/loadgen/LoadGen.java | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/src/main/java/com/github/wrossmorrow/extproc/loadgen/LoadGen.java b/src/main/java/com/github/wrossmorrow/extproc/loadgen/LoadGen.java index 04f9a4a..9c25378 100644 --- a/src/main/java/com/github/wrossmorrow/extproc/loadgen/LoadGen.java +++ b/src/main/java/com/github/wrossmorrow/extproc/loadgen/LoadGen.java @@ -172,7 +172,9 @@ static WorkerResult runWorker( result.errors.add(new ExchangeError(index, outcome.error)); } else { result.latencies.add(outcome.latency); - result.extprocDurationsNs.add(outcome.extprocDurationNs); + if (outcome.extprocDurationNs != null) { + result.extprocDurationsNs.add(outcome.extprocDurationNs); + } } index += stride; } From 859d446055c02b4b3a29c02a09ae8121ec36a641 Mon Sep 17 00:00:00 2001 From: "wrossmorrow-pi-bot[bot]" <346594324+wrossmorrow-pi-bot[bot]@users.noreply.github.com> Date: Wed, 28 Oct 2026 13:55:41 +0000 Subject: [PATCH 06/10] fix: await processing responses during load generator exchanges --- .../wrossmorrow/extproc/loadgen/LoadGen.java | 35 ++++++++++++++++++- 1 file changed, 34 insertions(+), 1 deletion(-) diff --git a/src/main/java/com/github/wrossmorrow/extproc/loadgen/LoadGen.java b/src/main/java/com/github/wrossmorrow/extproc/loadgen/LoadGen.java index 9c25378..235f6a2 100644 --- a/src/main/java/com/github/wrossmorrow/extproc/loadgen/LoadGen.java +++ b/src/main/java/com/github/wrossmorrow/extproc/loadgen/LoadGen.java @@ -17,6 +17,7 @@ import java.time.Instant; import java.util.ArrayList; import java.util.List; +import java.util.concurrent.CountDownLatch; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.Future; @@ -186,6 +187,8 @@ static ExchangeOutcome runExchange(ExternalProcessorGrpc.ExternalProcessorStub s ResponseCollector responseObserver = new ResponseCollector(); StreamObserver requestObserver = stub.process(responseObserver); + // phase 1: request headers (not yet end of stream) + responseObserver.latch = new CountDownLatch(1); requestObserver.onNext( ProcessingRequest.newBuilder() .setRequestHeaders( @@ -194,6 +197,11 @@ static ExchangeOutcome runExchange(ExternalProcessorGrpc.ExternalProcessorStub s .setEndOfStream(false) .build()) .build()); + awaitExchange(responseObserver); + ProcessingResponse response = responseObserver.lastResponse; + + // phase 2: request body (end of stream) + responseObserver.latch = new CountDownLatch(1); requestObserver.onNext( ProcessingRequest.newBuilder() .setRequestBody( @@ -202,10 +210,20 @@ static ExchangeOutcome runExchange(ExternalProcessorGrpc.ExternalProcessorStub s .setEndOfStream(true) .build()) .build()); + awaitExchange(responseObserver); + response = responseObserver.lastResponse; + + // phase 3: response headers (end of stream) + responseObserver.latch = new CountDownLatch(1); requestObserver.onNext( ProcessingRequest.newBuilder() .setResponseHeaders(HttpHeaders.newBuilder().setEndOfStream(true).build()) .build()); + awaitExchange(responseObserver); + response = responseObserver.lastResponse; + + // phase 4: response body (end of stream) + responseObserver.latch = new CountDownLatch(1); requestObserver.onNext( ProcessingRequest.newBuilder() .setResponseBody( @@ -215,13 +233,26 @@ static ExchangeOutcome runExchange(ExternalProcessorGrpc.ExternalProcessorStub s .build()) .build()); requestObserver.onCompleted(); + awaitExchange(responseObserver); + response = responseObserver.lastResponse; - Long extprocDurationNs = extractDurationNs(responseObserver.lastResponse); + Long extprocDurationNs = extractDurationNs(response); Duration latency = Duration.between(start, Instant.now()); String error = responseObserver.error == null ? null : responseObserver.error.getMessage(); return new ExchangeOutcome(latency, extprocDurationNs, error); } + static void awaitExchange(ResponseCollector responseObserver) { + try { + if (!responseObserver.latch.await(30, TimeUnit.SECONDS)) { + throw new RuntimeException("Timed out waiting for processing response"); + } + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new RuntimeException("Interrupted while waiting for processing response", e); + } + } + static Long extractDurationNs(ProcessingResponse response) { if (response == null) { return null; @@ -283,10 +314,12 @@ static final class ResponseCollector implements StreamObserver Date: Wed, 28 Oct 2026 14:12:56 +0000 Subject: [PATCH 07/10] fix: use plaintext transport for local extproc load generator connections --- .../java/com/github/wrossmorrow/extproc/loadgen/LoadGen.java | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/src/main/java/com/github/wrossmorrow/extproc/loadgen/LoadGen.java b/src/main/java/com/github/wrossmorrow/extproc/loadgen/LoadGen.java index 235f6a2..ec72ef9 100644 --- a/src/main/java/com/github/wrossmorrow/extproc/loadgen/LoadGen.java +++ b/src/main/java/com/github/wrossmorrow/extproc/loadgen/LoadGen.java @@ -82,8 +82,7 @@ public static void main(String[] args) throws Exception { host = addr.substring(0, colon); port = Integer.parseInt(addr.substring(colon + 1)); } - ManagedChannel channel = - ManagedChannelBuilder.forAddress(host, port).useTransportSecurity().build(); + ManagedChannel channel = ManagedChannelBuilder.forAddress(host, port).usePlaintext().build(); ExternalProcessorGrpc.ExternalProcessorStub stub = ExternalProcessorGrpc.newStub(channel); ExecutorService executor = Executors.newFixedThreadPool(concurrency); try { From f1b12f751af2fe1f368f458ce23ef66500fcc8fc Mon Sep 17 00:00:00 2001 From: "wrossmorrow-pi-bot[bot]" <346594324+wrossmorrow-pi-bot[bot]@users.noreply.github.com> Date: Wed, 28 Oct 2026 15:23:39 +0000 Subject: [PATCH 08/10] debug: add phase level tracing to load generator exchanges --- .../com/github/wrossmorrow/extproc/loadgen/LoadGen.java | 7 +++++++ 1 file changed, 7 insertions(+) diff --git a/src/main/java/com/github/wrossmorrow/extproc/loadgen/LoadGen.java b/src/main/java/com/github/wrossmorrow/extproc/loadgen/LoadGen.java index ec72ef9..a1ecdff 100644 --- a/src/main/java/com/github/wrossmorrow/extproc/loadgen/LoadGen.java +++ b/src/main/java/com/github/wrossmorrow/extproc/loadgen/LoadGen.java @@ -187,6 +187,7 @@ static ExchangeOutcome runExchange(ExternalProcessorGrpc.ExternalProcessorStub s StreamObserver requestObserver = stub.process(responseObserver); // phase 1: request headers (not yet end of stream) + System.out.println("LoadGen: exchange starting, sending phase 1"); responseObserver.latch = new CountDownLatch(1); requestObserver.onNext( ProcessingRequest.newBuilder() @@ -198,6 +199,7 @@ static ExchangeOutcome runExchange(ExternalProcessorGrpc.ExternalProcessorStub s .build()); awaitExchange(responseObserver); ProcessingResponse response = responseObserver.lastResponse; + System.out.println("LoadGen: got response for phase 1: " + response.getResponseCase()); // phase 2: request body (end of stream) responseObserver.latch = new CountDownLatch(1); @@ -211,6 +213,7 @@ static ExchangeOutcome runExchange(ExternalProcessorGrpc.ExternalProcessorStub s .build()); awaitExchange(responseObserver); response = responseObserver.lastResponse; + System.out.println("LoadGen: got response for phase 2: " + response.getResponseCase()); // phase 3: response headers (end of stream) responseObserver.latch = new CountDownLatch(1); @@ -220,6 +223,7 @@ static ExchangeOutcome runExchange(ExternalProcessorGrpc.ExternalProcessorStub s .build()); awaitExchange(responseObserver); response = responseObserver.lastResponse; + System.out.println("LoadGen: got response for phase 3: " + response.getResponseCase()); // phase 4: response body (end of stream) responseObserver.latch = new CountDownLatch(1); @@ -234,6 +238,7 @@ static ExchangeOutcome runExchange(ExternalProcessorGrpc.ExternalProcessorStub s requestObserver.onCompleted(); awaitExchange(responseObserver); response = responseObserver.lastResponse; + System.out.println("LoadGen: got response for phase 4: " + response.getResponseCase()); Long extprocDurationNs = extractDurationNs(response); Duration latency = Duration.between(start, Instant.now()); @@ -242,10 +247,12 @@ static ExchangeOutcome runExchange(ExternalProcessorGrpc.ExternalProcessorStub s } static void awaitExchange(ResponseCollector responseObserver) { + System.out.println("LoadGen: awaiting processing response"); try { if (!responseObserver.latch.await(30, TimeUnit.SECONDS)) { throw new RuntimeException("Timed out waiting for processing response"); } + System.out.println("LoadGen: got processing response"); } catch (InterruptedException e) { Thread.currentThread().interrupt(); throw new RuntimeException("Interrupted while waiting for processing response", e); From 3e5a4ae9c8c1ebe5ccf67c5703b44b0dea80a108 Mon Sep 17 00:00:00 2001 From: "wrossmorrow-pi-bot[bot]" <346594324+wrossmorrow-pi-bot[bot]@users.noreply.github.com> Date: Wed, 28 Oct 2026 15:35:59 +0000 Subject: [PATCH 09/10] infra: publish extproc echo port for local load testing --- docker-compose.yaml | 2 ++ 1 file changed, 2 insertions(+) diff --git a/docker-compose.yaml b/docker-compose.yaml index d458492..934f911 100644 --- a/docker-compose.yaml +++ b/docker-compose.yaml @@ -124,4 +124,6 @@ services: - -Dextproc.class=com.github.wrossmorrow.extproc.processors.EchoRequestProcessor - -jar - /extproc/extproc.jar + ports: + - "50051:50051" restart: always From 1de162602bdb4be8856e08af67abe5c6570fc13d Mon Sep 17 00:00:00 2001 From: "wrossmorrow-pi-bot[bot]" <346594324+wrossmorrow-pi-bot[bot]@users.noreply.github.com> Date: Wed, 28 Oct 2026 15:47:34 +0000 Subject: [PATCH 10/10] infra: revert example echo port exposure for load testing --- docker-compose.yaml | 2 -- 1 file changed, 2 deletions(-) diff --git a/docker-compose.yaml b/docker-compose.yaml index 934f911..d458492 100644 --- a/docker-compose.yaml +++ b/docker-compose.yaml @@ -124,6 +124,4 @@ services: - -Dextproc.class=com.github.wrossmorrow.extproc.processors.EchoRequestProcessor - -jar - /extproc/extproc.jar - ports: - - "50051:50051" restart: always