• Home
  • Features
  • Pricing
  • Docs
  • Announcements
  • Sign In

grpc / grpc-java / #20346

03 Jul 2026 10:23AM UTC coverage: 89.106% (-0.01%) from 89.118%
#20346

push

github

web-flow
xds: Move backend service label plumbing to CDS (#12882)

Addresses #12431.

This moves backend service metric label plumbing from `xds_cluster_impl`
to `cds`, matching the A75/A89 aggregate cluster behavior.

Previously, `xds_cluster_impl` added the `grpc.lb.backend_service`
pick-details label and propagated `NameResolver.ATTR_BACKEND_SERVICE` to
its child policy. That works for simple clusters, but it is the wrong
layer after A75 because aggregate clusters need the leaf CDS cluster
name, not the aggregate root cluster name.

This change makes CDS responsible for backend service plumbing:

- non-aggregate CDS adds the `grpc.lb.backend_service` pick-details
label using the CDS cluster name
- non-aggregate CDS propagates `NameResolver.ATTR_BACKEND_SERVICE` to
its child policy, so WRR can consume the backend service context
- aggregate root CDS does not add the backend service attribute or
pick-details label
- leaf CDS instances under an aggregate cluster add their own leaf
cluster name
- `xds_cluster_impl` no longer owns backend service pick-details label
or child attribute propagation

The existing `InternalEquivalentAddressGroup.ATTR_BACKEND_SERVICE` path
is left unchanged for endpoint/subchannel metrics.

Tests cover:

- non-aggregate CDS propagating `NameResolver.ATTR_BACKEND_SERVICE`
- non-aggregate CDS adding `grpc.lb.backend_service`
- aggregate root CDS not adding the root cluster name
- aggregate leaf CDS instances propagating leaf cluster names
- `xds_cluster_impl` no longer adding backend service labels or
attributes
- picker tests using a non-null `PickDetailsConsumer`, matching the real
`PickSubchannelArgs` contract

37992 of 42637 relevant lines covered (89.11%)

0.89 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

95.18
/../xds/src/main/java/io/grpc/xds/ClusterImplLoadBalancer.java
1
/*
2
 * Copyright 2020 The gRPC Authors
3
 *
4
 * Licensed under the Apache License, Version 2.0 (the "License");
5
 * you may not use this file except in compliance with the License.
6
 * You may obtain a copy of the License at
7
 *
8
 *     http://www.apache.org/licenses/LICENSE-2.0
9
 *
10
 * Unless required by applicable law or agreed to in writing, software
11
 * distributed under the License is distributed on an "AS IS" BASIS,
12
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13
 * See the License for the specific language governing permissions and
14
 * limitations under the License.
15
 */
16

17
package io.grpc.xds;
18

19
import static com.google.common.base.Preconditions.checkNotNull;
20
import static io.grpc.xds.client.LoadStatsManager2.isEnabledOrcaLrsPropagation;
21

22
import com.google.common.annotations.VisibleForTesting;
23
import com.google.common.base.MoreObjects;
24
import com.google.common.base.Strings;
25
import com.google.common.collect.ImmutableMap;
26
import com.google.protobuf.Struct;
27
import io.grpc.Attributes;
28
import io.grpc.ClientStreamTracer;
29
import io.grpc.ClientStreamTracer.StreamInfo;
30
import io.grpc.ConnectivityState;
31
import io.grpc.ConnectivityStateInfo;
32
import io.grpc.EquivalentAddressGroup;
33
import io.grpc.InternalLogId;
34
import io.grpc.LoadBalancer;
35
import io.grpc.Metadata;
36
import io.grpc.Status;
37
import io.grpc.internal.ForwardingClientStreamTracer;
38
import io.grpc.internal.GrpcUtil;
39
import io.grpc.services.MetricReport;
40
import io.grpc.util.ForwardingLoadBalancerHelper;
41
import io.grpc.util.ForwardingSubchannel;
42
import io.grpc.util.GracefulSwitchLoadBalancer;
43
import io.grpc.xds.ClusterImplLoadBalancerProvider.ClusterImplConfig;
44
import io.grpc.xds.Endpoints.DropOverload;
45
import io.grpc.xds.EnvoyServerProtoData.UpstreamTlsContext;
46
import io.grpc.xds.ThreadSafeRandom.ThreadSafeRandomImpl;
47
import io.grpc.xds.XdsNameResolverProvider.CallCounterProvider;
48
import io.grpc.xds.client.BackendMetricPropagation;
49
import io.grpc.xds.client.Bootstrapper.ServerInfo;
50
import io.grpc.xds.client.LoadStatsManager2.ClusterDropStats;
51
import io.grpc.xds.client.LoadStatsManager2.ClusterLocalityStats;
52
import io.grpc.xds.client.Locality;
53
import io.grpc.xds.client.XdsClient;
54
import io.grpc.xds.client.XdsLogger;
55
import io.grpc.xds.client.XdsLogger.XdsLogLevel;
56
import io.grpc.xds.internal.XdsInternalAttributes;
57
import io.grpc.xds.internal.security.SecurityProtocolNegotiators;
58
import io.grpc.xds.internal.security.SslContextProviderSupplier;
59
import io.grpc.xds.orca.OrcaPerRequestUtil;
60
import io.grpc.xds.orca.OrcaPerRequestUtil.OrcaPerRequestReportListener;
61
import java.util.ArrayList;
62
import java.util.Collections;
63
import java.util.List;
64
import java.util.Locale;
65
import java.util.Map;
66
import java.util.Objects;
67
import java.util.concurrent.atomic.AtomicLong;
68
import java.util.concurrent.atomic.AtomicReference;
69
import javax.annotation.Nullable;
70

71
/**
72
 * Load balancer for cluster_impl_experimental LB policy. This LB policy is the child LB policy of
73
 * the priority_experimental LB policy and the parent LB policy of the weighted_target_experimental
74
 * LB policy in the xDS load balancing hierarchy. This LB policy applies cluster-level
75
 * configurations to requests sent to the corresponding cluster, such as drop policies, circuit
76
 * breakers.
77
 */
78
final class ClusterImplLoadBalancer extends LoadBalancer {
79

80
  @VisibleForTesting
81
  static final long DEFAULT_PER_CLUSTER_MAX_CONCURRENT_REQUESTS = 1024L;
82
  @VisibleForTesting
83
  static boolean enableCircuitBreaking =
1✔
84
      Strings.isNullOrEmpty(System.getenv("GRPC_XDS_EXPERIMENTAL_CIRCUIT_BREAKING"))
1✔
85
          || Boolean.parseBoolean(System.getenv("GRPC_XDS_EXPERIMENTAL_CIRCUIT_BREAKING"));
1✔
86

87
  private static final Attributes.Key<AtomicReference<ClusterLocality>> ATTR_CLUSTER_LOCALITY =
1✔
88
      Attributes.Key.create("io.grpc.xds.ClusterImplLoadBalancer.clusterLocality");
1✔
89
  @VisibleForTesting
90
  static final Attributes.Key<String> ATTR_SUBCHANNEL_ADDRESS_NAME =
1✔
91
      Attributes.Key.create("io.grpc.xds.ClusterImplLoadBalancer.addressName");
1✔
92

93
  private final XdsLogger logger;
94
  private final Helper helper;
95
  private final ThreadSafeRandom random;
96
  // The following fields are effectively final.
97
  private String cluster;
98
  @Nullable
99
  private String edsServiceName;
100
  private XdsClient xdsClient;
101
  private CallCounterProvider callCounterProvider;
102
  private ClusterDropStats dropStats;
103
  private ClusterImplLbHelper childLbHelper;
104
  private GracefulSwitchLoadBalancer childSwitchLb;
105

106
  ClusterImplLoadBalancer(Helper helper) {
107
    this(helper, ThreadSafeRandomImpl.instance);
1✔
108
  }
1✔
109

110
  ClusterImplLoadBalancer(Helper helper, ThreadSafeRandom random) {
1✔
111
    this.helper = checkNotNull(helper, "helper");
1✔
112
    this.random = checkNotNull(random, "random");
1✔
113
    InternalLogId logId = InternalLogId.allocate("cluster-impl-lb", helper.getAuthority());
1✔
114
    logger = XdsLogger.withLogId(logId);
1✔
115
    logger.log(XdsLogLevel.INFO, "Created");
1✔
116
  }
1✔
117

118
  @Override
119
  public Status acceptResolvedAddresses(ResolvedAddresses resolvedAddresses) {
120
    logger.log(XdsLogLevel.DEBUG, "Received resolution result: {0}", resolvedAddresses);
1✔
121
    Attributes attributes = resolvedAddresses.getAttributes();
1✔
122
    if (xdsClient == null) {
1✔
123
      xdsClient = checkNotNull(attributes.get(io.grpc.xds.XdsAttributes.XDS_CLIENT), "xdsClient");
1✔
124
    }
125
    if (callCounterProvider == null) {
1✔
126
      callCounterProvider = attributes.get(io.grpc.xds.XdsAttributes.CALL_COUNTER_PROVIDER);
1✔
127
    }
128

129
    ClusterImplConfig config =
1✔
130
        (ClusterImplConfig) resolvedAddresses.getLoadBalancingPolicyConfig();
1✔
131
    if (config == null) {
1✔
132
      return Status.INTERNAL.withDescription("No cluster configuration found");
×
133
    }
134

135
    if (cluster == null) {
1✔
136
      cluster = config.cluster;
1✔
137
      edsServiceName = config.edsServiceName;
1✔
138
      childLbHelper = new ClusterImplLbHelper(
1✔
139
          callCounterProvider.getOrCreate(config.cluster, config.edsServiceName),
1✔
140
          config.lrsServerInfo);
141
      childSwitchLb = new GracefulSwitchLoadBalancer(childLbHelper);
1✔
142
      // Assume load report server does not change throughout cluster lifetime.
143
      if (config.lrsServerInfo != null) {
1✔
144
        dropStats = xdsClient.addClusterDropStats(config.lrsServerInfo, cluster, edsServiceName);
1✔
145
      }
146
    }
147

148
    childLbHelper.updateDropPolicies(config.dropCategories);
1✔
149
    childLbHelper.updateMaxConcurrentRequests(config.maxConcurrentRequests);
1✔
150
    childLbHelper.updateSslContextProviderSupplier(config.tlsContext);
1✔
151
    childLbHelper.updateFilterMetadata(config.filterMetadata);
1✔
152
    childLbHelper.updateBackendMetricPropagation(config.backendMetricPropagation);
1✔
153

154
    return childSwitchLb.acceptResolvedAddresses(
1✔
155
        resolvedAddresses.toBuilder()
1✔
156
            .setLoadBalancingPolicyConfig(config.childConfig)
1✔
157
            .build());
1✔
158
  }
159

160
  @Override
161
  public void handleNameResolutionError(Status error) {
162
    if (childSwitchLb != null) {
1✔
163
      childSwitchLb.handleNameResolutionError(error);
1✔
164
    } else {
165
      helper.updateBalancingState(
1✔
166
          ConnectivityState.TRANSIENT_FAILURE, new FixedResultPicker(PickResult.withError(error)));
1✔
167
    }
168
  }
1✔
169

170
  @Override
171
  public void requestConnection() {
172
    if (childSwitchLb != null) {
1✔
173
      childSwitchLb.requestConnection();
1✔
174
    }
175
  }
1✔
176

177
  @Override
178
  public void shutdown() {
179
    if (dropStats != null) {
1✔
180
      dropStats.release();
1✔
181
    }
182
    if (childSwitchLb != null) {
1✔
183
      childSwitchLb.shutdown();
1✔
184
      if (childLbHelper != null) {
1✔
185
        childLbHelper.updateSslContextProviderSupplier(null);
1✔
186
        childLbHelper = null;
1✔
187
      }
188
    }
189
    xdsClient = null;
1✔
190
  }
1✔
191

192
  /**
193
   * A decorated {@link LoadBalancer.Helper} that applies configurations for connections
194
   * or requests to endpoints in the cluster.
195
   */
196
  private final class ClusterImplLbHelper extends ForwardingLoadBalancerHelper {
197
    private final AtomicLong inFlights;
198
    private ConnectivityState currentState = ConnectivityState.IDLE;
1✔
199
    private SubchannelPicker currentPicker = new FixedResultPicker(PickResult.withNoResult());
1✔
200
    private List<DropOverload> dropPolicies = Collections.emptyList();
1✔
201
    private long maxConcurrentRequests = DEFAULT_PER_CLUSTER_MAX_CONCURRENT_REQUESTS;
1✔
202
    @Nullable
203
    private SslContextProviderSupplier sslContextProviderSupplier;
204
    private Map<String, Struct> filterMetadata = ImmutableMap.of();
1✔
205
    @Nullable
206
    private final ServerInfo lrsServerInfo;
207
    @Nullable
208
    private BackendMetricPropagation backendMetricPropagation;
209

210
    private ClusterImplLbHelper(AtomicLong inFlights, @Nullable ServerInfo lrsServerInfo) {
1✔
211
      this.inFlights = checkNotNull(inFlights, "inFlights");
1✔
212
      this.lrsServerInfo = lrsServerInfo;
1✔
213
    }
1✔
214

215
    @Override
216
    public void updateBalancingState(ConnectivityState newState, SubchannelPicker newPicker) {
217
      currentState = newState;
1✔
218
      currentPicker =  newPicker;
1✔
219
      SubchannelPicker picker = new RequestLimitingSubchannelPicker(
1✔
220
          newPicker, dropPolicies, maxConcurrentRequests, filterMetadata);
221
      delegate().updateBalancingState(newState, picker);
1✔
222
    }
1✔
223

224
    @Override
225
    public Subchannel createSubchannel(CreateSubchannelArgs args) {
226
      List<EquivalentAddressGroup> addresses = withAdditionalAttributes(args.getAddresses());
1✔
227
      // This value for  ClusterLocality is not recommended for general use.
228
      // Currently, we extract locality data from the first address, even before the subchannel is
229
      // READY.
230
      // This is mainly to accommodate scenarios where a Load Balancing API (like "pick first")
231
      // might return the subchannel before it is READY. Typically, we wouldn't report load for such
232
      // selections because the channel will disregard the chosen (not-ready) subchannel.
233
      // However, we needed to ensure this case is handled.
234
      ClusterLocality clusterLocality = createClusterLocalityFromAttributes(
1✔
235
          args.getAddresses().get(0).getAttributes());
1✔
236
      AtomicReference<ClusterLocality> localityAtomicReference = new AtomicReference<>(
1✔
237
          clusterLocality);
238
      Attributes.Builder attrsBuilder = args.getAttributes().toBuilder()
1✔
239
          .set(ATTR_CLUSTER_LOCALITY, localityAtomicReference);
1✔
240
      if (GrpcUtil.getFlag("GRPC_EXPERIMENTAL_XDS_AUTHORITY_REWRITE", false)) {
1✔
241
        String hostname = args.getAddresses().get(0).getAttributes()
1✔
242
            .get(XdsInternalAttributes.ATTR_ADDRESS_NAME);
1✔
243
        if (hostname != null) {
1✔
244
          attrsBuilder.set(ATTR_SUBCHANNEL_ADDRESS_NAME, hostname);
1✔
245
        }
246
      }
247
      args = args.toBuilder().setAddresses(addresses).setAttributes(attrsBuilder.build()).build();
1✔
248
      final Subchannel subchannel = delegate().createSubchannel(args);
1✔
249

250
      return new ClusterImplSubchannel(subchannel, localityAtomicReference);
1✔
251
    }
252

253
    private final class ClusterImplSubchannel extends ForwardingSubchannel {
254
      private final Subchannel delegate;
255
      private final AtomicReference<ClusterLocality> localityAtomicReference;
256

257
      private ClusterImplSubchannel(
258
          Subchannel delegate, AtomicReference<ClusterLocality> localityAtomicReference) {
1✔
259
        this.delegate = delegate;
1✔
260
        this.localityAtomicReference = localityAtomicReference;
1✔
261
      }
1✔
262

263
      @Override
264
      public void start(SubchannelStateListener listener) {
265
        delegate().start(
1✔
266
            new SubchannelStateListener() {
1✔
267
              @Override
268
              public void onSubchannelState(ConnectivityStateInfo newState) {
269
                // Do nothing if LB has been shutdown
270
                if (xdsClient != null && newState.getState().equals(ConnectivityState.READY)) {
1✔
271
                  // Get locality based on the connected address attributes
272
                  ClusterLocality updatedClusterLocality =
1✔
273
                      createClusterLocalityFromAttributes(
1✔
274
                          delegate.getConnectedAddressAttributes());
1✔
275
                  ClusterLocality oldClusterLocality =
1✔
276
                      localityAtomicReference.getAndSet(updatedClusterLocality);
1✔
277
                  oldClusterLocality.release();
1✔
278
                }
279
                listener.onSubchannelState(newState);
1✔
280
              }
1✔
281
            });
282
      }
1✔
283

284
      @Override
285
      public void shutdown() {
286
        localityAtomicReference.get().release();
1✔
287
        delegate().shutdown();
1✔
288
      }
1✔
289

290
      @Override
291
      public void updateAddresses(List<EquivalentAddressGroup> addresses) {
292
        delegate().updateAddresses(withAdditionalAttributes(addresses));
1✔
293
      }
1✔
294

295
      @Override
296
      protected Subchannel delegate() {
297
        return delegate;
1✔
298
      }
299
    }
300

301
    private List<EquivalentAddressGroup> withAdditionalAttributes(
302
        List<EquivalentAddressGroup> addresses) {
303
      List<EquivalentAddressGroup> newAddresses = new ArrayList<>();
1✔
304
      for (EquivalentAddressGroup eag : addresses) {
1✔
305
        Attributes.Builder attrBuilder = eag.getAttributes().toBuilder().set(
1✔
306
            io.grpc.xds.XdsAttributes.ATTR_CLUSTER_NAME, cluster);
1✔
307
        if (sslContextProviderSupplier != null) {
1✔
308
          attrBuilder.set(
1✔
309
              SecurityProtocolNegotiators.ATTR_SSL_CONTEXT_PROVIDER_SUPPLIER,
310
              sslContextProviderSupplier);
311
        }
312
        newAddresses.add(new EquivalentAddressGroup(eag.getAddresses(), attrBuilder.build()));
1✔
313
      }
1✔
314
      return newAddresses;
1✔
315
    }
316

317
    private ClusterLocality createClusterLocalityFromAttributes(Attributes addressAttributes) {
318
      Locality locality = addressAttributes.get(io.grpc.xds.XdsAttributes.ATTR_LOCALITY);
1✔
319
      String localityName = addressAttributes.get(EquivalentAddressGroup.ATTR_LOCALITY_NAME);
1✔
320

321
      // Endpoint addresses resolved by ClusterResolverLoadBalancer should always contain
322
      // attributes with its locality, including endpoints in LOGICAL_DNS clusters.
323
      // In case of not (which really shouldn't), loads are aggregated under an empty
324
      // locality.
325
      if (locality == null) {
1✔
326
        locality = Locality.create("", "", "");
×
327
        localityName = "";
×
328
      }
329

330
      final ClusterLocalityStats localityStats =
331
          (lrsServerInfo == null)
1✔
332
              ? null
1✔
333
              : xdsClient.addClusterLocalityStats(lrsServerInfo, cluster,
1✔
334
                  edsServiceName, locality, backendMetricPropagation);
1✔
335

336
      return new ClusterLocality(localityStats, localityName);
1✔
337
    }
338

339
    @Override
340
    protected Helper delegate()  {
341
      return helper;
1✔
342
    }
343

344
    private void updateDropPolicies(List<DropOverload> dropOverloads) {
345
      if (!dropPolicies.equals(dropOverloads)) {
1✔
346
        dropPolicies = dropOverloads;
1✔
347
        updateBalancingState(currentState, currentPicker);
1✔
348
      }
349
    }
1✔
350

351
    private void updateMaxConcurrentRequests(@Nullable Long maxConcurrentRequests) {
352
      if (Objects.equals(this.maxConcurrentRequests, maxConcurrentRequests)) {
1✔
353
        return;
×
354
      }
355
      this.maxConcurrentRequests =
1✔
356
          maxConcurrentRequests != null
1✔
357
              ? maxConcurrentRequests
1✔
358
              : DEFAULT_PER_CLUSTER_MAX_CONCURRENT_REQUESTS;
1✔
359
      updateBalancingState(currentState, currentPicker);
1✔
360
    }
1✔
361

362
    private void updateSslContextProviderSupplier(@Nullable UpstreamTlsContext tlsContext) {
363
      UpstreamTlsContext currentTlsContext =
364
          sslContextProviderSupplier != null
1✔
365
              ? (UpstreamTlsContext)sslContextProviderSupplier.getTlsContext()
1✔
366
              : null;
1✔
367
      if (Objects.equals(currentTlsContext,  tlsContext)) {
1✔
368
        return;
1✔
369
      }
370
      if (sslContextProviderSupplier != null) {
1✔
371
        sslContextProviderSupplier.close();
1✔
372
      }
373
      sslContextProviderSupplier =
1✔
374
          tlsContext != null
1✔
375
              ? new SslContextProviderSupplier(tlsContext,
1✔
376
                                               (TlsContextManager) xdsClient.getSecurityConfig())
1✔
377
              : null;
1✔
378
    }
1✔
379

380
    private void updateFilterMetadata(Map<String, Struct> filterMetadata) {
381
      this.filterMetadata = ImmutableMap.copyOf(filterMetadata);
1✔
382
    }
1✔
383

384
    private void updateBackendMetricPropagation(
385
        @Nullable BackendMetricPropagation backendMetricPropagation) {
386
      this.backendMetricPropagation = backendMetricPropagation;
1✔
387
    }
1✔
388

389
    private class RequestLimitingSubchannelPicker extends SubchannelPicker {
390
      private final SubchannelPicker delegate;
391
      private final List<DropOverload> dropPolicies;
392
      private final long maxConcurrentRequests;
393
      private final Map<String, Struct> filterMetadata;
394

395
      private RequestLimitingSubchannelPicker(SubchannelPicker delegate,
396
          List<DropOverload> dropPolicies, long maxConcurrentRequests,
397
          Map<String, Struct> filterMetadata) {
1✔
398
        this.delegate = delegate;
1✔
399
        this.dropPolicies = dropPolicies;
1✔
400
        this.maxConcurrentRequests = maxConcurrentRequests;
1✔
401
        this.filterMetadata = checkNotNull(filterMetadata, "filterMetadata");
1✔
402
      }
1✔
403

404
      @Override
405
      public PickResult pickSubchannel(PickSubchannelArgs args) {
406
        args.getCallOptions().getOption(ClusterImplLoadBalancerProvider.FILTER_METADATA_CONSUMER)
1✔
407
            .accept(filterMetadata);
1✔
408
        for (DropOverload dropOverload : dropPolicies) {
1✔
409
          int rand = random.nextInt(1_000_000);
1✔
410
          if (rand < dropOverload.dropsPerMillion()) {
1✔
411
            logger.log(XdsLogLevel.INFO, "Drop request with category: {0}",
1✔
412
                dropOverload.category());
1✔
413
            if (dropStats != null) {
1✔
414
              dropStats.recordDroppedRequest(dropOverload.category());
1✔
415
            }
416
            return PickResult.withDrop(
1✔
417
                Status.UNAVAILABLE.withDescription("Dropped: " + dropOverload.category()));
1✔
418
          }
419
        }
1✔
420
        PickResult result = delegate.pickSubchannel(args);
1✔
421
        if (result.getStatus().isOk() && result.getSubchannel() != null) {
1✔
422
          Subchannel subchannel = result.getSubchannel();
1✔
423
          if (subchannel instanceof ClusterImplLbHelper.ClusterImplSubchannel) {
1✔
424
            subchannel = ((ClusterImplLbHelper.ClusterImplSubchannel) subchannel).delegate();
1✔
425
            result = result.copyWithSubchannel(subchannel);
1✔
426
          }
427
          if (enableCircuitBreaking) {
1✔
428
            if (inFlights.get() >= maxConcurrentRequests) {
1✔
429
              if (dropStats != null) {
1✔
430
                dropStats.recordDroppedRequest();
1✔
431
              }
432
              return PickResult.withDrop(Status.UNAVAILABLE.withDescription(
1✔
433
                  String.format(Locale.US, "Cluster max concurrent requests limit of %d exceeded",
1✔
434
                      maxConcurrentRequests)));
1✔
435
            }
436
          }
437
          final AtomicReference<ClusterLocality> clusterLocality =
1✔
438
              result.getSubchannel().getAttributes().get(ATTR_CLUSTER_LOCALITY);
1✔
439

440
          if (clusterLocality != null) {
1✔
441
            ClusterLocalityStats stats = clusterLocality.get().getClusterLocalityStats();
1✔
442
            if (stats != null) {
1✔
443
              String localityName =
1✔
444
                  result.getSubchannel().getAttributes().get(ATTR_CLUSTER_LOCALITY).get()
1✔
445
                      .getClusterLocalityName();
1✔
446
              args.getPickDetailsConsumer().addOptionalLabel("grpc.lb.locality", localityName);
1✔
447

448
              ClientStreamTracer.Factory tracerFactory = new CountingStreamTracerFactory(
1✔
449
                  stats, inFlights, result.getStreamTracerFactory());
1✔
450
              ClientStreamTracer.Factory orcaTracerFactory = OrcaPerRequestUtil.getInstance()
1✔
451
                  .newOrcaClientStreamTracerFactory(tracerFactory, new OrcaPerRpcListener(stats));
1✔
452
              result = result.copyWithStreamTracerFactory(orcaTracerFactory);
1✔
453
            }
454
          }
455
          if (args.getCallOptions().getOption(XdsNameResolver.AUTO_HOST_REWRITE_KEY) != null
1✔
456
              && args.getCallOptions().getOption(XdsNameResolver.AUTO_HOST_REWRITE_KEY)) {
1✔
457
            result = PickResult.withSubchannel(result.getSubchannel(),
1✔
458
                result.getStreamTracerFactory(),
1✔
459
                result.getSubchannel().getAttributes().get(ATTR_SUBCHANNEL_ADDRESS_NAME));
1✔
460
          }
461
        }
462
        return result;
1✔
463
      }
464

465
      @Override
466
      public String toString() {
467
        return MoreObjects.toStringHelper(this).add("delegate", delegate).toString();
×
468
      }
469
    }
470
  }
471

472
  private static final class CountingStreamTracerFactory extends
473
      ClientStreamTracer.Factory {
474
    private final ClusterLocalityStats stats;
475
    private final AtomicLong inFlights;
476
    @Nullable
477
    private final ClientStreamTracer.Factory delegate;
478

479
    private CountingStreamTracerFactory(
480
        ClusterLocalityStats stats, AtomicLong inFlights,
481
        @Nullable ClientStreamTracer.Factory delegate) {
1✔
482
      this.stats = checkNotNull(stats, "stats");
1✔
483
      this.inFlights = checkNotNull(inFlights, "inFlights");
1✔
484
      this.delegate = delegate;
1✔
485
    }
1✔
486

487
    @Override
488
    public ClientStreamTracer newClientStreamTracer(StreamInfo info, Metadata headers) {
489
      stats.recordCallStarted();
1✔
490
      inFlights.incrementAndGet();
1✔
491
      if (delegate == null) {
1✔
492
        return new ClientStreamTracer() {
1✔
493
          @Override
494
          public void streamClosed(Status status) {
495
            stats.recordCallFinished(status);
1✔
496
            inFlights.decrementAndGet();
1✔
497
          }
1✔
498
        };
499
      }
500
      final ClientStreamTracer delegatedTracer = delegate.newClientStreamTracer(info, headers);
×
501
      return new ForwardingClientStreamTracer() {
×
502
        @Override
503
        protected ClientStreamTracer delegate() {
504
          return delegatedTracer;
×
505
        }
506

507
        @Override
508
        public void streamClosed(Status status) {
509
          stats.recordCallFinished(status);
×
510
          inFlights.decrementAndGet();
×
511
          delegate().streamClosed(status);
×
512
        }
×
513
      };
514
    }
515
  }
516

517
  private static final class OrcaPerRpcListener implements OrcaPerRequestReportListener {
518

519
    private final ClusterLocalityStats stats;
520

521
    private OrcaPerRpcListener(ClusterLocalityStats stats) {
1✔
522
      this.stats = checkNotNull(stats, "stats");
1✔
523
    }
1✔
524

525
    /**
526
     * Copies ORCA metrics from {@link MetricReport} to {@link ClusterLocalityStats}
527
     * such that they are included in the snapshot for the LRS report sent to the LRS server.
528
     * This includes both top-level metrics (CPU, memory, application utilization) and named
529
     * metrics, filtered according to the backend metric propagation configuration.
530
     */
531
    @Override
532
    public void onLoadReport(MetricReport report) {
533
      if (isEnabledOrcaLrsPropagation) {
1✔
534
        stats.recordTopLevelMetrics(
1✔
535
            report.getCpuUtilization(),
1✔
536
            report.getMemoryUtilization(),
1✔
537
            report.getApplicationUtilization());
1✔
538
      }
539
      stats.recordBackendLoadMetricStats(report.getNamedMetrics());
1✔
540
    }
1✔
541
  }
542

543
  /**
544
   * Represents the {@link ClusterLocalityStats} and network locality name of a cluster.
545
   */
546
  static final class ClusterLocality {
547
    private final ClusterLocalityStats clusterLocalityStats;
548
    private final String clusterLocalityName;
549

550
    @VisibleForTesting
551
    ClusterLocality(ClusterLocalityStats localityStats, String localityName) {
1✔
552
      this.clusterLocalityStats = localityStats;
1✔
553
      this.clusterLocalityName = localityName;
1✔
554
    }
1✔
555

556
    ClusterLocalityStats getClusterLocalityStats() {
557
      return clusterLocalityStats;
1✔
558
    }
559

560
    String getClusterLocalityName() {
561
      return clusterLocalityName;
1✔
562
    }
563

564
    @VisibleForTesting
565
    void release() {
566
      if (clusterLocalityStats != null) {
1✔
567
        clusterLocalityStats.release();
1✔
568
      }
569
    }
1✔
570
  }
571
}
STATUS · Troubleshooting · Open an Issue · Sales · Support · CAREERS · ENTERPRISE · START FREE · SCHEDULE DEMO
ANNOUNCEMENTS · TWITTER · TOS & SLA · Supported CI Services · What's a CI service? · Automated Testing

© 2026 Coveralls, Inc