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

Implementing dropping backend_service metrics label, and the discrepancy
with the gRFC in `end_of_stream_without_message` handling. The code
treats this field independently, but as per the gRFC it should be only
be considered when the `end_of_stream` field is true. This applies both
in `ProcessingRequest` and `ProcessingResponse`.

Author: kannanjgithub

xds/src/main/java/io/grpc/xds/ExternalProcessorClientInterceptor.java

@@ -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(
@@ -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(
@@ -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(
@@ -162,7 +162,7 @@ static synchronized void initMetricInstruments() {
             "s",
             LATENCY_BUCKETS,
             ImmutableList.of("grpc.target"),
-            ImmutableList.of("grpc.lb.backend_service"),
+            ImmutableList.of(),
             true);
       }
     }
@@ -256,8 +256,7 @@ public void cancel(@Nullable String message, @Nullable Throwable cause) {
 
     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;
   }
@@ -350,7 +349,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;
@@ -383,8 +381,7 @@ protected DataPlaneClientCall(
         MethodDescriptor<?, ?> method,
         Channel channel,
         MetricRecorder metricsRecorder,
-        String target,
-        String backendService) {
+        String target) {
       super(delayedCall);
       this.delayedCall = delayedCall;
       this.rawCall = rawCall;
@@ -397,7 +394,6 @@ protected DataPlaneClientCall(
       this.channel = channel;
       this.metricsRecorder = checkNotNull(metricsRecorder, "metricsRecorder");
       this.target = checkNotNull(target, "target");
-      this.backendService = checkNotNull(backendService, "backendService");
     }
 
     private boolean activateCall() {
@@ -429,7 +425,7 @@ private void recordDuration(DoubleHistogramMetricInstrument instrument, long dur
             instrument,
             durationSecs,
             ImmutableList.of(target),
-            ImmutableList.of(backendService));
+            ImmutableList.of());
       }
     }
 
@@ -1134,6 +1130,7 @@ public void halfClose() {
 
         ProcessingRequest.Builder builder = ProcessingRequest.newBuilder()
             .setRequestBody(HttpBody.newBuilder()
+                .setEndOfStream(true)
                 .setEndOfStreamWithoutMessage(true)
                 .build());
         mergeAccumulatedWindowUpdates(builder);
@@ -1167,7 +1164,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) {
@@ -1184,7 +1184,7 @@ private void handleRequestBodyResponse(BodyResponse bodyResponse) {
               trySendAccumulatedWindowUpdates();
             }
           }
-          if (streamed.getEndOfStream() || streamed.getEndOfStreamWithoutMessage()) {
+          if (isEndOfStream) {
             synchronized (streamLock) {
               if (pendingUpstreamBodyMessages.isEmpty()) {
                 if (requestSideClosed.compareAndSet(false, true)) {