Skip to content

xds: Ext proc client drop backend_service metrics label, make eos_without_message conditional on eos field #13028

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
kannanjgithub wants to merge 6 commits into
base: master
Choose a base branch
from
Open

xds: Ext proc client drop backend_service metrics label, make eos_without_message conditional on eos field #13028

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
9366622
xds: improve test coverage for ExternalProcessorClientInterceptor
kannanjgithub Aug 21, 2026
6505b99
xds: drop backend_service metric label from ext_proc client metrics
kannanjgithub Sep 4, 2026
61b6fec
xds: set end_of_stream on client half-close and handle end_of_stream_…
kannanjgithub Sep 4, 2026
f93f5df
Merge remote-tracking branch 'origin/master' into ext-proc-client
kannanjgithub Sep 4, 2026
6d2335d
xds: redo ext_proc client half-close and metric label changes after m…
kannanjgithub Sep 4, 2026
7233b3c
xds: fix checkstyle violations in ExternalProcessorClientInterceptorTest
kannanjgithub Sep 4, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
26 changes: 13 additions & 13 deletions xds/src/main/java/io/grpc/xds/ExternalProcessorClientInterceptor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -132,7 +132,7 @@ static synchronized void initMetricInstruments() {
"s",
LATENCY_BUCKETS,
ImmutableList.of("grpc.target"),
ImmutableList.of("grpc.lb.backend_service"),
ImmutableList.of(),
true);

clientHalfCloseDuration = registry.registerDoubleHistogram(
Expand All @@ -142,7 +142,7 @@ static synchronized void initMetricInstruments() {
"s",
LATENCY_BUCKETS,
ImmutableList.of("grpc.target"),
ImmutableList.of("grpc.lb.backend_service"),
ImmutableList.of(),
true);

serverHeadersDuration = registry.registerDoubleHistogram(
Expand All @@ -152,7 +152,7 @@ static synchronized void initMetricInstruments() {
"s",
LATENCY_BUCKETS,
ImmutableList.of("grpc.target"),
ImmutableList.of("grpc.lb.backend_service"),
ImmutableList.of(),
true);

serverTrailersDuration = registry.registerDoubleHistogram(
Expand All @@ -162,7 +162,7 @@ static synchronized void initMetricInstruments() {
"s",
LATENCY_BUCKETS,
ImmutableList.of("grpc.target"),
ImmutableList.of("grpc.lb.backend_service"),
ImmutableList.of(),
true);
}
}
Expand Down Expand Up @@ -246,8 +246,7 @@ public <ReqT, RespT> ClientCall<ReqT, RespT> interceptCall(

DataPlaneClientCall dataPlaneCall = new DataPlaneClientCall(
delayedCall, rawCall, extProcStub, filterConfig, filterConfig.getMutationRulesConfig(),
scheduler, rawMethod, next, metricsRecorder, next.authority(),
callOptions.getOption(XdsNameResolver.CLUSTER_SELECTION_KEY));
scheduler, rawMethod, next, metricsRecorder, next.authority());

return (ClientCall<ReqT, RespT>) (ClientCall<?, ?>) dataPlaneCall;
}
Expand Down Expand Up @@ -340,7 +339,6 @@ private static class DataPlaneClientCall
private final Channel channel;
private final MetricRecorder metricsRecorder;
private final String target;
private final String backendService;
private volatile Context callContext = Context.ROOT;

private volatile long clientHeadersStartNanos;
Expand Down Expand Up @@ -374,8 +372,7 @@ protected DataPlaneClientCall(
MethodDescriptor<?, ?> method,
Channel channel,
MetricRecorder metricsRecorder,
String target,
String backendService) {
String target) {
super(delayedCall);
this.delayedCall = delayedCall;
this.rawCall = rawCall;
Expand All @@ -388,7 +385,6 @@ protected DataPlaneClientCall(
this.channel = channel;
this.metricsRecorder = checkNotNull(metricsRecorder, "metricsRecorder");
this.target = checkNotNull(target, "target");
this.backendService = checkNotNull(backendService, "backendService");
}

private boolean activateCall() {
Expand Down Expand Up @@ -420,7 +416,7 @@ private void recordDuration(DoubleHistogramMetricInstrument instrument, long dur
instrument,
durationSecs,
ImmutableList.of(target),
ImmutableList.of(backendService));
ImmutableList.of());
}
}

Expand Down Expand Up @@ -1124,6 +1120,7 @@ public void halfClose() {

ProcessingRequest.Builder builder = ProcessingRequest.newBuilder()
.setRequestBody(HttpBody.newBuilder()
.setEndOfStream(true)
.setEndOfStreamWithoutMessage(true)
.build());
mergeAccumulatedWindowUpdates(builder);
Expand Down Expand Up @@ -1156,7 +1153,10 @@ private void handleRequestBodyResponse(BodyResponse bodyResponse) {
BodyMutation mutation = bodyResponse.getResponse().getBodyMutation();
if (mutation.hasStreamedResponse()) {
StreamedBodyResponse streamed = mutation.getStreamedResponse();
if (!streamed.getEndOfStreamWithoutMessage()) {
boolean isEndOfStream = streamed.getEndOfStream();
boolean isEndOfStreamWithoutMessage =
isEndOfStream && streamed.getEndOfStreamWithoutMessage();
if (!isEndOfStreamWithoutMessage) {
ByteString body = streamed.getBody();
boolean sendImmediately = false;
synchronized (streamLock) {
Expand All @@ -1173,7 +1173,7 @@ private void handleRequestBodyResponse(BodyResponse bodyResponse) {
trySendAccumulatedWindowUpdates();
}
}
if (streamed.getEndOfStream() || streamed.getEndOfStreamWithoutMessage()) {
if (isEndOfStream) {
synchronized (streamLock) {
if (pendingUpstreamBodyMessages.isEmpty()) {
if (requestSideClosed.compareAndSet(false, true)) {
Expand Down
Loading
Loading