xds: redo ext_proc client half-close and metric label changes after master merge

- Drop backend_service metric label from ExternalProcessorClientInterceptor.
- Set end_of_stream on client half-close alongside end_of_stream_without_message.
- Handle end_of_stream_without_message per ext_proc spec on StreamedBodyResponse.
- Update tests and mocks to reflect the updated half-close semantics.

TAG=agy
CONV=86c74ebe-9fd7-4876-b27c-a4c1b230d346

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);
       }
     }
@@ -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;
   }
@@ -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;
@@ -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;
@@ -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() {
@@ -420,7 +416,7 @@ private void recordDuration(DoubleHistogramMetricInstrument instrument, long dur
             instrument,
             durationSecs,
             ImmutableList.of(target),
-            ImmutableList.of(backendService));
+            ImmutableList.of());
       }
     }
 
@@ -1124,6 +1120,7 @@ public void halfClose() {
 
         ProcessingRequest.Builder builder = ProcessingRequest.newBuilder()
             .setRequestBody(HttpBody.newBuilder()
+                .setEndOfStream(true)
                 .setEndOfStreamWithoutMessage(true)
                 .build());
         mergeAccumulatedWindowUpdates(builder);
@@ -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) {
@@ -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)) {