#19471

Committed 25 Sep 2024 05:16PM CUT coverage: 84.511% (-0.02%) from 84.529%

Build # #19471

Build Type

push

github

Committed by

web-flow

Commit Message

xds: Check for validity of xdsClient in ClusterImplLbHelper (#11553) (#11554)

* Added null check for xdsClient in onSubChannelState. This avoids NPE
for xdsClient when LB is shutdown and onSubChannelState is called later
as part of listener callback. As shutdown is racy and eventually consistent,
this check would avoid calculating locality after LB is shutdown.

Run Details

33535 of 39681 relevant lines covered (84.51%)

0.85 hits per line

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

94.55

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

/*
 * Copyright 2020 The gRPC Authors
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package io.grpc.xds;

import static com.google.common.base.Preconditions.checkNotNull;

import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.MoreObjects;
import com.google.common.base.Strings;
import com.google.common.collect.ImmutableMap;
import com.google.protobuf.Struct;
import io.grpc.Attributes;
import io.grpc.ClientStreamTracer;
import io.grpc.ClientStreamTracer.StreamInfo;
import io.grpc.ConnectivityState;
import io.grpc.ConnectivityStateInfo;
import io.grpc.EquivalentAddressGroup;
import io.grpc.InternalLogId;
import io.grpc.LoadBalancer;
import io.grpc.Metadata;
import io.grpc.Status;
import io.grpc.internal.ForwardingClientStreamTracer;
import io.grpc.internal.ObjectPool;
import io.grpc.services.MetricReport;
import io.grpc.util.ForwardingLoadBalancerHelper;
import io.grpc.util.ForwardingSubchannel;
import io.grpc.util.GracefulSwitchLoadBalancer;
import io.grpc.xds.ClusterImplLoadBalancerProvider.ClusterImplConfig;
import io.grpc.xds.Endpoints.DropOverload;
import io.grpc.xds.EnvoyServerProtoData.UpstreamTlsContext;
import io.grpc.xds.ThreadSafeRandom.ThreadSafeRandomImpl;
import io.grpc.xds.XdsNameResolverProvider.CallCounterProvider;
import io.grpc.xds.client.Bootstrapper.ServerInfo;
import io.grpc.xds.client.LoadStatsManager2.ClusterDropStats;
import io.grpc.xds.client.LoadStatsManager2.ClusterLocalityStats;
import io.grpc.xds.client.Locality;
import io.grpc.xds.client.XdsClient;
import io.grpc.xds.client.XdsLogger;
import io.grpc.xds.client.XdsLogger.XdsLogLevel;
import io.grpc.xds.internal.security.SslContextProviderSupplier;
import io.grpc.xds.orca.OrcaPerRequestUtil;
import io.grpc.xds.orca.OrcaPerRequestUtil.OrcaPerRequestReportListener;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;
import javax.annotation.Nullable;

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

  @VisibleForTesting
  static final long DEFAULT_PER_CLUSTER_MAX_CONCURRENT_REQUESTS = 1024L;
  @VisibleForTesting
  static boolean enableCircuitBreaking =
      Strings.isNullOrEmpty(System.getenv("GRPC_XDS_EXPERIMENTAL_CIRCUIT_BREAKING"))
          || Boolean.parseBoolean(System.getenv("GRPC_XDS_EXPERIMENTAL_CIRCUIT_BREAKING"));

  private static final Attributes.Key<AtomicReference<ClusterLocality>> ATTR_CLUSTER_LOCALITY =
      Attributes.Key.create("io.grpc.xds.ClusterImplLoadBalancer.clusterLocality");

  private final XdsLogger logger;
  private final Helper helper;
  private final ThreadSafeRandom random;
  // The following fields are effectively final.
  private String cluster;
  @Nullable
  private String edsServiceName;
  private ObjectPool<XdsClient> xdsClientPool;
  private XdsClient xdsClient;
  private CallCounterProvider callCounterProvider;
  private ClusterDropStats dropStats;
  private ClusterImplLbHelper childLbHelper;
  private GracefulSwitchLoadBalancer childSwitchLb;

  ClusterImplLoadBalancer(Helper helper) {
    this(helper, ThreadSafeRandomImpl.instance);
  }

  ClusterImplLoadBalancer(Helper helper, ThreadSafeRandom random) {
    this.helper = checkNotNull(helper, "helper");
    this.random = checkNotNull(random, "random");
    InternalLogId logId = InternalLogId.allocate("cluster-impl-lb", helper.getAuthority());
    logger = XdsLogger.withLogId(logId);
    logger.log(XdsLogLevel.INFO, "Created");
  }

  @Override
  public Status acceptResolvedAddresses(ResolvedAddresses resolvedAddresses) {
    logger.log(XdsLogLevel.DEBUG, "Received resolution result: {0}", resolvedAddresses);
    Attributes attributes = resolvedAddresses.getAttributes();
    if (xdsClientPool == null) {
      xdsClientPool = attributes.get(InternalXdsAttributes.XDS_CLIENT_POOL);
      assert xdsClientPool != null;
      xdsClient = xdsClientPool.getObject();
    }
    if (callCounterProvider == null) {
      callCounterProvider = attributes.get(InternalXdsAttributes.CALL_COUNTER_PROVIDER);
    }

    ClusterImplConfig config =
        (ClusterImplConfig) resolvedAddresses.getLoadBalancingPolicyConfig();
    if (config == null) {
      return Status.INTERNAL.withDescription("No cluster configuration found");
    }

    if (cluster == null) {
      cluster = config.cluster;
      edsServiceName = config.edsServiceName;
      childLbHelper = new ClusterImplLbHelper(
          callCounterProvider.getOrCreate(config.cluster, config.edsServiceName),
          config.lrsServerInfo);
      childSwitchLb = new GracefulSwitchLoadBalancer(childLbHelper);
      // Assume load report server does not change throughout cluster lifetime.
      if (config.lrsServerInfo != null) {
        dropStats = xdsClient.addClusterDropStats(config.lrsServerInfo, cluster, edsServiceName);
      }
    }

    childLbHelper.updateDropPolicies(config.dropCategories);
    childLbHelper.updateMaxConcurrentRequests(config.maxConcurrentRequests);
    childLbHelper.updateSslContextProviderSupplier(config.tlsContext);
    childLbHelper.updateFilterMetadata(config.filterMetadata);

    childSwitchLb.handleResolvedAddresses(
        resolvedAddresses.toBuilder()
            .setAttributes(attributes)
            .setLoadBalancingPolicyConfig(config.childConfig)
            .build());
    return Status.OK;
  }

  @Override
  public void handleNameResolutionError(Status error) {
    if (childSwitchLb != null) {
      childSwitchLb.handleNameResolutionError(error);
    } else {
      helper.updateBalancingState(
          ConnectivityState.TRANSIENT_FAILURE, new FixedResultPicker(PickResult.withError(error)));
    }
  }

  @Override
  public void shutdown() {
    if (dropStats != null) {
      dropStats.release();
    }
    if (childSwitchLb != null) {
      childSwitchLb.shutdown();
      if (childLbHelper != null) {
        childLbHelper.updateSslContextProviderSupplier(null);
        childLbHelper = null;
      }
    }
    if (xdsClient != null) {
      xdsClient = xdsClientPool.returnObject(xdsClient);
    }
  }

  /**
   * A decorated {@link LoadBalancer.Helper} that applies configurations for connections
   * or requests to endpoints in the cluster.
   */
  private final class ClusterImplLbHelper extends ForwardingLoadBalancerHelper {
    private final AtomicLong inFlights;
    private ConnectivityState currentState = ConnectivityState.IDLE;
    private SubchannelPicker currentPicker = new FixedResultPicker(PickResult.withNoResult());
    private List<DropOverload> dropPolicies = Collections.emptyList();
    private long maxConcurrentRequests = DEFAULT_PER_CLUSTER_MAX_CONCURRENT_REQUESTS;
    @Nullable
    private SslContextProviderSupplier sslContextProviderSupplier;
    private Map<String, Struct> filterMetadata = ImmutableMap.of();
    @Nullable
    private final ServerInfo lrsServerInfo;

    private ClusterImplLbHelper(AtomicLong inFlights, @Nullable ServerInfo lrsServerInfo) {
      this.inFlights = checkNotNull(inFlights, "inFlights");
      this.lrsServerInfo = lrsServerInfo;
    }

    @Override
    public void updateBalancingState(ConnectivityState newState, SubchannelPicker newPicker) {
      currentState = newState;
      currentPicker =  newPicker;
      SubchannelPicker picker = new RequestLimitingSubchannelPicker(
          newPicker, dropPolicies, maxConcurrentRequests, filterMetadata);
      delegate().updateBalancingState(newState, picker);
    }

    @Override
    public Subchannel createSubchannel(CreateSubchannelArgs args) {
      List<EquivalentAddressGroup> addresses = withAdditionalAttributes(args.getAddresses());
      // This value for  ClusterLocality is not recommended for general use.
      // Currently, we extract locality data from the first address, even before the subchannel is
      // READY.
      // This is mainly to accommodate scenarios where a Load Balancing API (like "pick first")
      // might return the subchannel before it is READY. Typically, we wouldn't report load for such
      // selections because the channel will disregard the chosen (not-ready) subchannel.
      // However, we needed to ensure this case is handled.
      ClusterLocality clusterLocality = createClusterLocalityFromAttributes(
          args.getAddresses().get(0).getAttributes());
      AtomicReference<ClusterLocality> localityAtomicReference = new AtomicReference<>(
          clusterLocality);
      Attributes attrs = args.getAttributes().toBuilder()
          .set(ATTR_CLUSTER_LOCALITY, localityAtomicReference)
          .build();
      args = args.toBuilder().setAddresses(addresses).setAttributes(attrs).build();
      final Subchannel subchannel = delegate().createSubchannel(args);

      return new ForwardingSubchannel() {
        @Override
        public void start(SubchannelStateListener listener) {
          delegate().start(new SubchannelStateListener() {
            @Override
            public void onSubchannelState(ConnectivityStateInfo newState) {
              // Do nothing if LB has been shutdown
              if (xdsClient != null && newState.getState().equals(ConnectivityState.READY)) {
                // Get locality based on the connected address attributes
                ClusterLocality updatedClusterLocality = createClusterLocalityFromAttributes(
                    subchannel.getConnectedAddressAttributes());
                ClusterLocality oldClusterLocality = localityAtomicReference
                    .getAndSet(updatedClusterLocality);
                oldClusterLocality.release();
              }
              listener.onSubchannelState(newState);
            }
          });
        }

        @Override
        public void shutdown() {
          localityAtomicReference.get().release();
          delegate().shutdown();
        }

        @Override
        public void updateAddresses(List<EquivalentAddressGroup> addresses) {
          delegate().updateAddresses(withAdditionalAttributes(addresses));
        }

        @Override
        protected Subchannel delegate() {
          return subchannel;
        }
      };
    }

    private List<EquivalentAddressGroup> withAdditionalAttributes(
        List<EquivalentAddressGroup> addresses) {
      List<EquivalentAddressGroup> newAddresses = new ArrayList<>();
      for (EquivalentAddressGroup eag : addresses) {
        Attributes.Builder attrBuilder = eag.getAttributes().toBuilder().set(
            InternalXdsAttributes.ATTR_CLUSTER_NAME, cluster);
        if (sslContextProviderSupplier != null) {
          attrBuilder.set(
              InternalXdsAttributes.ATTR_SSL_CONTEXT_PROVIDER_SUPPLIER,
              sslContextProviderSupplier);
        }
        newAddresses.add(new EquivalentAddressGroup(eag.getAddresses(), attrBuilder.build()));
      }
      return newAddresses;
    }

    private ClusterLocality createClusterLocalityFromAttributes(Attributes addressAttributes) {
      Locality locality = addressAttributes.get(InternalXdsAttributes.ATTR_LOCALITY);
      String localityName = addressAttributes.get(InternalXdsAttributes.ATTR_LOCALITY_NAME);

      // Endpoint addresses resolved by ClusterResolverLoadBalancer should always contain
      // attributes with its locality, including endpoints in LOGICAL_DNS clusters.
      // In case of not (which really shouldn't), loads are aggregated under an empty
      // locality.
      if (locality == null) {
        locality = Locality.create("", "", "");
        localityName = "";
      }

      final ClusterLocalityStats localityStats =
          (lrsServerInfo == null)
              ? null
              : xdsClient.addClusterLocalityStats(lrsServerInfo, cluster,
                  edsServiceName, locality);

      return new ClusterLocality(localityStats, localityName);
    }

    @Override
    protected Helper delegate()  {
      return helper;
    }

    private void updateDropPolicies(List<DropOverload> dropOverloads) {
      if (!dropPolicies.equals(dropOverloads)) {
        dropPolicies = dropOverloads;
        updateBalancingState(currentState, currentPicker);
      }
    }

    private void updateMaxConcurrentRequests(@Nullable Long maxConcurrentRequests) {
      if (Objects.equals(this.maxConcurrentRequests, maxConcurrentRequests)) {
        return;
      }
      this.maxConcurrentRequests =
          maxConcurrentRequests != null
              ? maxConcurrentRequests
              : DEFAULT_PER_CLUSTER_MAX_CONCURRENT_REQUESTS;
      updateBalancingState(currentState, currentPicker);
    }

    private void updateSslContextProviderSupplier(@Nullable UpstreamTlsContext tlsContext) {
      UpstreamTlsContext currentTlsContext =
          sslContextProviderSupplier != null
              ? (UpstreamTlsContext)sslContextProviderSupplier.getTlsContext()
              : null;
      if (Objects.equals(currentTlsContext,  tlsContext)) {
        return;
      }
      if (sslContextProviderSupplier != null) {
        sslContextProviderSupplier.close();
      }
      sslContextProviderSupplier =
          tlsContext != null
              ? new SslContextProviderSupplier(tlsContext,
                                               (TlsContextManager) xdsClient.getSecurityConfig())
              : null;
    }

    private void updateFilterMetadata(Map<String, Struct> filterMetadata) {
      this.filterMetadata = ImmutableMap.copyOf(filterMetadata);
    }

    private class RequestLimitingSubchannelPicker extends SubchannelPicker {
      private final SubchannelPicker delegate;
      private final List<DropOverload> dropPolicies;
      private final long maxConcurrentRequests;
      private final Map<String, Struct> filterMetadata;

      private RequestLimitingSubchannelPicker(SubchannelPicker delegate,
          List<DropOverload> dropPolicies, long maxConcurrentRequests,
          Map<String, Struct> filterMetadata) {
        this.delegate = delegate;
        this.dropPolicies = dropPolicies;
        this.maxConcurrentRequests = maxConcurrentRequests;
        this.filterMetadata = checkNotNull(filterMetadata, "filterMetadata");
      }

      @Override
      public PickResult pickSubchannel(PickSubchannelArgs args) {
        args.getCallOptions().getOption(ClusterImplLoadBalancerProvider.FILTER_METADATA_CONSUMER)
            .accept(filterMetadata);
        for (DropOverload dropOverload : dropPolicies) {
          int rand = random.nextInt(1_000_000);
          if (rand < dropOverload.dropsPerMillion()) {
            logger.log(XdsLogLevel.INFO, "Drop request with category: {0}",
                dropOverload.category());
            if (dropStats != null) {
              dropStats.recordDroppedRequest(dropOverload.category());
            }
            return PickResult.withDrop(
                Status.UNAVAILABLE.withDescription("Dropped: " + dropOverload.category()));
          }
        }
        final PickResult result = delegate.pickSubchannel(args);
        if (result.getStatus().isOk() && result.getSubchannel() != null) {
          if (enableCircuitBreaking) {
            if (inFlights.get() >= maxConcurrentRequests) {
              if (dropStats != null) {
                dropStats.recordDroppedRequest();
              }
              return PickResult.withDrop(Status.UNAVAILABLE.withDescription(
                  "Cluster max concurrent requests limit exceeded"));
            }
          }
          final AtomicReference<ClusterLocality> clusterLocality =
              result.getSubchannel().getAttributes().get(ATTR_CLUSTER_LOCALITY);

          if (clusterLocality != null) {
            ClusterLocalityStats stats = clusterLocality.get().getClusterLocalityStats();
            if (stats != null) {
              String localityName =
                  result.getSubchannel().getAttributes().get(ATTR_CLUSTER_LOCALITY).get()
                      .getClusterLocalityName();
              args.getPickDetailsConsumer().addOptionalLabel("grpc.lb.locality", localityName);

              ClientStreamTracer.Factory tracerFactory = new CountingStreamTracerFactory(
                  stats, inFlights, result.getStreamTracerFactory());
              ClientStreamTracer.Factory orcaTracerFactory = OrcaPerRequestUtil.getInstance()
                  .newOrcaClientStreamTracerFactory(tracerFactory, new OrcaPerRpcListener(stats));
              return PickResult.withSubchannel(result.getSubchannel(), orcaTracerFactory);
            }
          }
        }
        return result;
      }

      @Override
      public String toString() {
        return MoreObjects.toStringHelper(this).add("delegate", delegate).toString();
      }
    }
  }

  private static final class CountingStreamTracerFactory extends
      ClientStreamTracer.Factory {
    private final ClusterLocalityStats stats;
    private final AtomicLong inFlights;
    @Nullable
    private final ClientStreamTracer.Factory delegate;

    private CountingStreamTracerFactory(
        ClusterLocalityStats stats, AtomicLong inFlights,
        @Nullable ClientStreamTracer.Factory delegate) {
      this.stats = checkNotNull(stats, "stats");
      this.inFlights = checkNotNull(inFlights, "inFlights");
      this.delegate = delegate;
    }

    @Override
    public ClientStreamTracer newClientStreamTracer(StreamInfo info, Metadata headers) {
      stats.recordCallStarted();
      inFlights.incrementAndGet();
      if (delegate == null) {
        return new ClientStreamTracer() {
          @Override
          public void streamClosed(Status status) {
            stats.recordCallFinished(status);
            inFlights.decrementAndGet();
          }
        };
      }
      final ClientStreamTracer delegatedTracer = delegate.newClientStreamTracer(info, headers);
      return new ForwardingClientStreamTracer() {
        @Override
        protected ClientStreamTracer delegate() {
          return delegatedTracer;
        }

        @Override
        public void streamClosed(Status status) {
          stats.recordCallFinished(status);
          inFlights.decrementAndGet();
          delegate().streamClosed(status);
        }
      };
    }
  }

  private static final class OrcaPerRpcListener implements OrcaPerRequestReportListener {

    private final ClusterLocalityStats stats;

    private OrcaPerRpcListener(ClusterLocalityStats stats) {
      this.stats = checkNotNull(stats, "stats");
    }

    /**
     * Copies {@link MetricReport#getNamedMetrics()} to {@link ClusterLocalityStats} such that it is
     * included in the snapshot for the LRS report sent to the LRS server.
     */
    @Override
    public void onLoadReport(MetricReport report) {
      stats.recordBackendLoadMetricStats(report.getNamedMetrics());
    }
  }

  /**
   * Represents the {@link ClusterLocalityStats} and network locality name of a cluster.
   */
  static final class ClusterLocality {
    private final ClusterLocalityStats clusterLocalityStats;
    private final String clusterLocalityName;

    @VisibleForTesting
    ClusterLocality(ClusterLocalityStats localityStats, String localityName) {
      this.clusterLocalityStats = localityStats;
      this.clusterLocalityName = localityName;
    }

    ClusterLocalityStats getClusterLocalityStats() {
      return clusterLocalityStats;
    }

    String getClusterLocalityName() {
      return clusterLocalityName;
    }

    @VisibleForTesting
    void release() {
      if (clusterLocalityStats != null) {
        clusterLocalityStats.release();
      }
    }
  }
}

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
21	import com.google.common.annotations.VisibleForTesting;
22	import com.google.common.base.MoreObjects;
23	import com.google.common.base.Strings;
24	import com.google.common.collect.ImmutableMap;
25	import com.google.protobuf.Struct;
26	import io.grpc.Attributes;
27	import io.grpc.ClientStreamTracer;
28	import io.grpc.ClientStreamTracer.StreamInfo;
29	import io.grpc.ConnectivityState;
30	import io.grpc.ConnectivityStateInfo;
31	import io.grpc.EquivalentAddressGroup;
32	import io.grpc.InternalLogId;
33	import io.grpc.LoadBalancer;
34	import io.grpc.Metadata;
35	import io.grpc.Status;
36	import io.grpc.internal.ForwardingClientStreamTracer;
37	import io.grpc.internal.ObjectPool;
38	import io.grpc.services.MetricReport;
39	import io.grpc.util.ForwardingLoadBalancerHelper;
40	import io.grpc.util.ForwardingSubchannel;
41	import io.grpc.util.GracefulSwitchLoadBalancer;
42	import io.grpc.xds.ClusterImplLoadBalancerProvider.ClusterImplConfig;
43	import io.grpc.xds.Endpoints.DropOverload;
44	import io.grpc.xds.EnvoyServerProtoData.UpstreamTlsContext;
45	import io.grpc.xds.ThreadSafeRandom.ThreadSafeRandomImpl;
46	import io.grpc.xds.XdsNameResolverProvider.CallCounterProvider;
47	import io.grpc.xds.client.Bootstrapper.ServerInfo;
48	import io.grpc.xds.client.LoadStatsManager2.ClusterDropStats;
49	import io.grpc.xds.client.LoadStatsManager2.ClusterLocalityStats;
50	import io.grpc.xds.client.Locality;
51	import io.grpc.xds.client.XdsClient;
52	import io.grpc.xds.client.XdsLogger;
53	import io.grpc.xds.client.XdsLogger.XdsLogLevel;
54	import io.grpc.xds.internal.security.SslContextProviderSupplier;
55	import io.grpc.xds.orca.OrcaPerRequestUtil;
56	import io.grpc.xds.orca.OrcaPerRequestUtil.OrcaPerRequestReportListener;
57	import java.util.ArrayList;
58	import java.util.Collections;
59	import java.util.List;
60	import java.util.Map;
61	import java.util.Objects;
62	import java.util.concurrent.atomic.AtomicLong;
63	import java.util.concurrent.atomic.AtomicReference;
64	import javax.annotation.Nullable;
65
66	/**
67	* Load balancer for cluster_impl_experimental LB policy. This LB policy is the child LB policy of
68	* the priority_experimental LB policy and the parent LB policy of the weighted_target_experimental
69	* LB policy in the xDS load balancing hierarchy. This LB policy applies cluster-level
70	* configurations to requests sent to the corresponding cluster, such as drop policies, circuit
71	* breakers.
72	*/
73	final class ClusterImplLoadBalancer extends LoadBalancer {
74
75	@VisibleForTesting
76	static final long DEFAULT_PER_CLUSTER_MAX_CONCURRENT_REQUESTS = 1024L;
77	@VisibleForTesting
78	static boolean enableCircuitBreaking =	1✔
79	Strings.isNullOrEmpty(System.getenv("GRPC_XDS_EXPERIMENTAL_CIRCUIT_BREAKING"))	1✔
80	\|\| Boolean.parseBoolean(System.getenv("GRPC_XDS_EXPERIMENTAL_CIRCUIT_BREAKING"));	1✔
81
82	private static final Attributes.Key<AtomicReference<ClusterLocality>> ATTR_CLUSTER_LOCALITY =	1✔
83	Attributes.Key.create("io.grpc.xds.ClusterImplLoadBalancer.clusterLocality");	1✔
84
85	private final XdsLogger logger;
86	private final Helper helper;
87	private final ThreadSafeRandom random;
88	// The following fields are effectively final.
89	private String cluster;
90	@Nullable
91	private String edsServiceName;
92	private ObjectPool<XdsClient> xdsClientPool;
93	private XdsClient xdsClient;
94	private CallCounterProvider callCounterProvider;
95	private ClusterDropStats dropStats;
96	private ClusterImplLbHelper childLbHelper;
97	private GracefulSwitchLoadBalancer childSwitchLb;
98
99	ClusterImplLoadBalancer(Helper helper) {
100	this(helper, ThreadSafeRandomImpl.instance);	1✔
101	}	1✔
102
103	ClusterImplLoadBalancer(Helper helper, ThreadSafeRandom random) {	1✔
104	this.helper = checkNotNull(helper, "helper");	1✔
105	this.random = checkNotNull(random, "random");	1✔
106	InternalLogId logId = InternalLogId.allocate("cluster-impl-lb", helper.getAuthority());	1✔
107	logger = XdsLogger.withLogId(logId);	1✔
108	logger.log(XdsLogLevel.INFO, "Created");	1✔
109	}	1✔
110
111	@Override
112	public Status acceptResolvedAddresses(ResolvedAddresses resolvedAddresses) {
113	logger.log(XdsLogLevel.DEBUG, "Received resolution result: {0}", resolvedAddresses);	1✔
114	Attributes attributes = resolvedAddresses.getAttributes();	1✔
115	if (xdsClientPool == null) {	1✔
116	xdsClientPool = attributes.get(InternalXdsAttributes.XDS_CLIENT_POOL);	1✔
117	assert xdsClientPool != null;	1✔
118	xdsClient = xdsClientPool.getObject();	1✔
119	}
120	if (callCounterProvider == null) {	1✔
121	callCounterProvider = attributes.get(InternalXdsAttributes.CALL_COUNTER_PROVIDER);	1✔
122	}
123
124	ClusterImplConfig config =	1✔
125	(ClusterImplConfig) resolvedAddresses.getLoadBalancingPolicyConfig();	1✔
126	if (config == null) {	1✔
127	return Status.INTERNAL.withDescription("No cluster configuration found");	×
128	}
129
130	if (cluster == null) {	1✔
131	cluster = config.cluster;	1✔
132	edsServiceName = config.edsServiceName;	1✔
133	childLbHelper = new ClusterImplLbHelper(	1✔
134	callCounterProvider.getOrCreate(config.cluster, config.edsServiceName),	1✔
135	config.lrsServerInfo);
136	childSwitchLb = new GracefulSwitchLoadBalancer(childLbHelper);	1✔
137	// Assume load report server does not change throughout cluster lifetime.
138	if (config.lrsServerInfo != null) {	1✔
139	dropStats = xdsClient.addClusterDropStats(config.lrsServerInfo, cluster, edsServiceName);	1✔
140	}
141	}
142
143	childLbHelper.updateDropPolicies(config.dropCategories);	1✔
144	childLbHelper.updateMaxConcurrentRequests(config.maxConcurrentRequests);	1✔
145	childLbHelper.updateSslContextProviderSupplier(config.tlsContext);	1✔
146	childLbHelper.updateFilterMetadata(config.filterMetadata);	1✔
147
148	childSwitchLb.handleResolvedAddresses(	1✔
149	resolvedAddresses.toBuilder()	1✔
150	.setAttributes(attributes)	1✔
151	.setLoadBalancingPolicyConfig(config.childConfig)	1✔
152	.build());	1✔
153	return Status.OK;	1✔
154	}
155
156	@Override
157	public void handleNameResolutionError(Status error) {
158	if (childSwitchLb != null) {	1✔
159	childSwitchLb.handleNameResolutionError(error);	1✔
160	} else {
161	helper.updateBalancingState(	1✔
162	ConnectivityState.TRANSIENT_FAILURE, new FixedResultPicker(PickResult.withError(error)));	1✔
163	}
164	}	1✔
165
166	@Override
167	public void shutdown() {
168	if (dropStats != null) {	1✔
169	dropStats.release();	1✔
170	}
171	if (childSwitchLb != null) {	1✔
172	childSwitchLb.shutdown();	1✔
173	if (childLbHelper != null) {	1✔
174	childLbHelper.updateSslContextProviderSupplier(null);	1✔
175	childLbHelper = null;	1✔
176	}
177	}
178	if (xdsClient != null) {	1✔
179	xdsClient = xdsClientPool.returnObject(xdsClient);	1✔
180	}
181	}	1✔
182
183	/**
184	* A decorated {@link LoadBalancer.Helper} that applies configurations for connections
185	* or requests to endpoints in the cluster.
186	*/
187	private final class ClusterImplLbHelper extends ForwardingLoadBalancerHelper {
188	private final AtomicLong inFlights;
189	private ConnectivityState currentState = ConnectivityState.IDLE;	1✔
190	private SubchannelPicker currentPicker = new FixedResultPicker(PickResult.withNoResult());	1✔
191	private List<DropOverload> dropPolicies = Collections.emptyList();	1✔
192	private long maxConcurrentRequests = DEFAULT_PER_CLUSTER_MAX_CONCURRENT_REQUESTS;	1✔
193	@Nullable
194	private SslContextProviderSupplier sslContextProviderSupplier;
195	private Map<String, Struct> filterMetadata = ImmutableMap.of();	1✔
196	@Nullable
197	private final ServerInfo lrsServerInfo;
198
199	private ClusterImplLbHelper(AtomicLong inFlights, @Nullable ServerInfo lrsServerInfo) {	1✔
200	this.inFlights = checkNotNull(inFlights, "inFlights");	1✔
201	this.lrsServerInfo = lrsServerInfo;	1✔
202	}	1✔
203
204	@Override
205	public void updateBalancingState(ConnectivityState newState, SubchannelPicker newPicker) {
206	currentState = newState;	1✔
207	currentPicker = newPicker;	1✔
208	SubchannelPicker picker = new RequestLimitingSubchannelPicker(	1✔
209	newPicker, dropPolicies, maxConcurrentRequests, filterMetadata);
210	delegate().updateBalancingState(newState, picker);	1✔
211	}	1✔
212
213	@Override
214	public Subchannel createSubchannel(CreateSubchannelArgs args) {
215	List<EquivalentAddressGroup> addresses = withAdditionalAttributes(args.getAddresses());	1✔
216	// This value for ClusterLocality is not recommended for general use.
217	// Currently, we extract locality data from the first address, even before the subchannel is
218	// READY.
219	// This is mainly to accommodate scenarios where a Load Balancing API (like "pick first")
220	// might return the subchannel before it is READY. Typically, we wouldn't report load for such
221	// selections because the channel will disregard the chosen (not-ready) subchannel.
222	// However, we needed to ensure this case is handled.
223	ClusterLocality clusterLocality = createClusterLocalityFromAttributes(	1✔
224	args.getAddresses().get(0).getAttributes());	1✔
225	AtomicReference<ClusterLocality> localityAtomicReference = new AtomicReference<>(	1✔
226	clusterLocality);
227	Attributes attrs = args.getAttributes().toBuilder()	1✔
228	.set(ATTR_CLUSTER_LOCALITY, localityAtomicReference)	1✔
229	.build();	1✔
230	args = args.toBuilder().setAddresses(addresses).setAttributes(attrs).build();	1✔
231	final Subchannel subchannel = delegate().createSubchannel(args);	1✔
232
233	return new ForwardingSubchannel() {	1✔
234	@Override
235	public void start(SubchannelStateListener listener) {
236	delegate().start(new SubchannelStateListener() {	1✔
237	@Override
238	public void onSubchannelState(ConnectivityStateInfo newState) {
239	// Do nothing if LB has been shutdown
240	if (xdsClient != null && newState.getState().equals(ConnectivityState.READY)) {	1✔
241	// Get locality based on the connected address attributes
242	ClusterLocality updatedClusterLocality = createClusterLocalityFromAttributes(	1✔
243	subchannel.getConnectedAddressAttributes());	1✔
244	ClusterLocality oldClusterLocality = localityAtomicReference	1✔
245	.getAndSet(updatedClusterLocality);	1✔
246	oldClusterLocality.release();	1✔
247	}
248	listener.onSubchannelState(newState);	1✔
249	}	1✔
250	});
251	}	1✔
252
253	@Override
254	public void shutdown() {
255	localityAtomicReference.get().release();	1✔
256	delegate().shutdown();	1✔
257	}	1✔
258
259	@Override
260	public void updateAddresses(List<EquivalentAddressGroup> addresses) {
261	delegate().updateAddresses(withAdditionalAttributes(addresses));	1✔
262	}	1✔
263
264	@Override
265	protected Subchannel delegate() {
266	return subchannel;	1✔
267	}
268	};
269	}
270
271	private List<EquivalentAddressGroup> withAdditionalAttributes(
272	List<EquivalentAddressGroup> addresses) {
273	List<EquivalentAddressGroup> newAddresses = new ArrayList<>();	1✔
274	for (EquivalentAddressGroup eag : addresses) {	1✔
275	Attributes.Builder attrBuilder = eag.getAttributes().toBuilder().set(	1✔
276	InternalXdsAttributes.ATTR_CLUSTER_NAME, cluster);	1✔
277	if (sslContextProviderSupplier != null) {	1✔
278	attrBuilder.set(	1✔
279	InternalXdsAttributes.ATTR_SSL_CONTEXT_PROVIDER_SUPPLIER,
280	sslContextProviderSupplier);
281	}
282	newAddresses.add(new EquivalentAddressGroup(eag.getAddresses(), attrBuilder.build()));	1✔
283	}	1✔
284	return newAddresses;	1✔
285	}
286
287	private ClusterLocality createClusterLocalityFromAttributes(Attributes addressAttributes) {
288	Locality locality = addressAttributes.get(InternalXdsAttributes.ATTR_LOCALITY);	1✔
289	String localityName = addressAttributes.get(InternalXdsAttributes.ATTR_LOCALITY_NAME);	1✔
290
291	// Endpoint addresses resolved by ClusterResolverLoadBalancer should always contain
292	// attributes with its locality, including endpoints in LOGICAL_DNS clusters.
293	// In case of not (which really shouldn't), loads are aggregated under an empty
294	// locality.
295	if (locality == null) {	1✔
296	locality = Locality.create("", "", "");	×
297	localityName = "";	×
298	}
299
300	final ClusterLocalityStats localityStats =
301	(lrsServerInfo == null)	1✔
302	? null	1✔
303	: xdsClient.addClusterLocalityStats(lrsServerInfo, cluster,	1✔
304	edsServiceName, locality);	1✔
305
306	return new ClusterLocality(localityStats, localityName);	1✔
307	}
308
309	@Override
310	protected Helper delegate() {
311	return helper;	1✔
312	}
313
314	private void updateDropPolicies(List<DropOverload> dropOverloads) {
315	if (!dropPolicies.equals(dropOverloads)) {	1✔
316	dropPolicies = dropOverloads;	1✔
317	updateBalancingState(currentState, currentPicker);	1✔
318	}
319	}	1✔
320
321	private void updateMaxConcurrentRequests(@Nullable Long maxConcurrentRequests) {
322	if (Objects.equals(this.maxConcurrentRequests, maxConcurrentRequests)) {	1✔
323	return;	×
324	}
325	this.maxConcurrentRequests =	1✔
326	maxConcurrentRequests != null	1✔
327	? maxConcurrentRequests	1✔
328	: DEFAULT_PER_CLUSTER_MAX_CONCURRENT_REQUESTS;	1✔
329	updateBalancingState(currentState, currentPicker);	1✔
330	}	1✔
331
332	private void updateSslContextProviderSupplier(@Nullable UpstreamTlsContext tlsContext) {
333	UpstreamTlsContext currentTlsContext =
334	sslContextProviderSupplier != null	1✔
335	? (UpstreamTlsContext)sslContextProviderSupplier.getTlsContext()	1✔
336	: null;	1✔
337	if (Objects.equals(currentTlsContext, tlsContext)) {	1✔
338	return;	1✔
339	}
340	if (sslContextProviderSupplier != null) {	1✔
341	sslContextProviderSupplier.close();	1✔
342	}
343	sslContextProviderSupplier =	1✔
344	tlsContext != null	1✔
345	? new SslContextProviderSupplier(tlsContext,	1✔
346	(TlsContextManager) xdsClient.getSecurityConfig())	1✔
347	: null;	1✔
348	}	1✔
349
350	private void updateFilterMetadata(Map<String, Struct> filterMetadata) {
351	this.filterMetadata = ImmutableMap.copyOf(filterMetadata);	1✔
352	}	1✔
353
354	private class RequestLimitingSubchannelPicker extends SubchannelPicker {
355	private final SubchannelPicker delegate;
356	private final List<DropOverload> dropPolicies;
357	private final long maxConcurrentRequests;
358	private final Map<String, Struct> filterMetadata;
359
360	private RequestLimitingSubchannelPicker(SubchannelPicker delegate,
361	List<DropOverload> dropPolicies, long maxConcurrentRequests,
362	Map<String, Struct> filterMetadata) {	1✔
363	this.delegate = delegate;	1✔
364	this.dropPolicies = dropPolicies;	1✔
365	this.maxConcurrentRequests = maxConcurrentRequests;	1✔
366	this.filterMetadata = checkNotNull(filterMetadata, "filterMetadata");	1✔
367	}	1✔
368
369	@Override
370	public PickResult pickSubchannel(PickSubchannelArgs args) {
371	args.getCallOptions().getOption(ClusterImplLoadBalancerProvider.FILTER_METADATA_CONSUMER)	1✔
372	.accept(filterMetadata);	1✔
373	for (DropOverload dropOverload : dropPolicies) {	1✔
374	int rand = random.nextInt(1_000_000);	1✔
375	if (rand < dropOverload.dropsPerMillion()) {	1✔
376	logger.log(XdsLogLevel.INFO, "Drop request with category: {0}",	1✔
377	dropOverload.category());	1✔
378	if (dropStats != null) {	1✔
379	dropStats.recordDroppedRequest(dropOverload.category());	1✔
380	}
381	return PickResult.withDrop(	1✔
382	Status.UNAVAILABLE.withDescription("Dropped: " + dropOverload.category()));	1✔
383	}
384	}	1✔
385	final PickResult result = delegate.pickSubchannel(args);	1✔
386	if (result.getStatus().isOk() && result.getSubchannel() != null) {	1✔
387	if (enableCircuitBreaking) {	1✔
388	if (inFlights.get() >= maxConcurrentRequests) {	1✔
389	if (dropStats != null) {	1✔
390	dropStats.recordDroppedRequest();	1✔
391	}
392	return PickResult.withDrop(Status.UNAVAILABLE.withDescription(	1✔
393	"Cluster max concurrent requests limit exceeded"));
394	}
395	}
396	final AtomicReference<ClusterLocality> clusterLocality =	1✔
397	result.getSubchannel().getAttributes().get(ATTR_CLUSTER_LOCALITY);	1✔
398
399	if (clusterLocality != null) {	1✔
400	ClusterLocalityStats stats = clusterLocality.get().getClusterLocalityStats();	1✔
401	if (stats != null) {	1✔
402	String localityName =	1✔
403	result.getSubchannel().getAttributes().get(ATTR_CLUSTER_LOCALITY).get()	1✔
404	.getClusterLocalityName();	1✔
405	args.getPickDetailsConsumer().addOptionalLabel("grpc.lb.locality", localityName);	1✔
406
407	ClientStreamTracer.Factory tracerFactory = new CountingStreamTracerFactory(	1✔
408	stats, inFlights, result.getStreamTracerFactory());	1✔
409	ClientStreamTracer.Factory orcaTracerFactory = OrcaPerRequestUtil.getInstance()	1✔
410	.newOrcaClientStreamTracerFactory(tracerFactory, new OrcaPerRpcListener(stats));	1✔
411	return PickResult.withSubchannel(result.getSubchannel(), orcaTracerFactory);	1✔
412	}
413	}
414	}
415	return result;	1✔
416	}
417
418	@Override
419	public String toString() {
420	return MoreObjects.toStringHelper(this).add("delegate", delegate).toString();	×
421	}
422	}
423	}
424
425	private static final class CountingStreamTracerFactory extends
426	ClientStreamTracer.Factory {
427	private final ClusterLocalityStats stats;
428	private final AtomicLong inFlights;
429	@Nullable
430	private final ClientStreamTracer.Factory delegate;
431
432	private CountingStreamTracerFactory(
433	ClusterLocalityStats stats, AtomicLong inFlights,
434	@Nullable ClientStreamTracer.Factory delegate) {	1✔
435	this.stats = checkNotNull(stats, "stats");	1✔
436	this.inFlights = checkNotNull(inFlights, "inFlights");	1✔
437	this.delegate = delegate;	1✔
438	}	1✔
439
440	@Override
441	public ClientStreamTracer newClientStreamTracer(StreamInfo info, Metadata headers) {
442	stats.recordCallStarted();	1✔
443	inFlights.incrementAndGet();	1✔
444	if (delegate == null) {	1✔
445	return new ClientStreamTracer() {	1✔
446	@Override
447	public void streamClosed(Status status) {
448	stats.recordCallFinished(status);	1✔
449	inFlights.decrementAndGet();	1✔
450	}	1✔
451	};
452	}
453	final ClientStreamTracer delegatedTracer = delegate.newClientStreamTracer(info, headers);	×
454	return new ForwardingClientStreamTracer() {	×
455	@Override
456	protected ClientStreamTracer delegate() {
457	return delegatedTracer;	×
458	}
459
460	@Override
461	public void streamClosed(Status status) {
462	stats.recordCallFinished(status);	×
463	inFlights.decrementAndGet();	×
464	delegate().streamClosed(status);	×
465	}	×
466	};
467	}
468	}
469
470	private static final class OrcaPerRpcListener implements OrcaPerRequestReportListener {
471
472	private final ClusterLocalityStats stats;
473
474	private OrcaPerRpcListener(ClusterLocalityStats stats) {	1✔
475	this.stats = checkNotNull(stats, "stats");	1✔
476	}	1✔
477
478	/**
479	* Copies {@link MetricReport#getNamedMetrics()} to {@link ClusterLocalityStats} such that it is
480	* included in the snapshot for the LRS report sent to the LRS server.
481	*/
482	@Override
483	public void onLoadReport(MetricReport report) {
484	stats.recordBackendLoadMetricStats(report.getNamedMetrics());	1✔
485	}	1✔
486	}
487
488	/**
489	* Represents the {@link ClusterLocalityStats} and network locality name of a cluster.
490	*/
491	static final class ClusterLocality {
492	private final ClusterLocalityStats clusterLocalityStats;
493	private final String clusterLocalityName;
494
495	@VisibleForTesting
496	ClusterLocality(ClusterLocalityStats localityStats, String localityName) {	1✔
497	this.clusterLocalityStats = localityStats;	1✔
498	this.clusterLocalityName = localityName;	1✔
499	}	1✔
500
501	ClusterLocalityStats getClusterLocalityStats() {
502	return clusterLocalityStats;	1✔
503	}
504
505	String getClusterLocalityName() {
506	return clusterLocalityName;	1✔
507	}
508
509	@VisibleForTesting
510	void release() {
511	if (clusterLocalityStats != null) {	1✔
512	clusterLocalityStats.release();	1✔
513	}
514	}	1✔
515	}
516	}

grpc / grpc-java / #19471

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

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