#19472

Committed 25 Sep 2024 06:11PM UTC coverage: 84.55% (+0.006%) from 84.544%

Build # #19472

Build Type

push

github

Committed by

ejona86

Commit Message

xds: Improve ClusterImpl's FakeSubchannel to verify state changes

The main goal was to make sure subchannels went CONNECTING only after a
connection was requested (since the test doesn't transition to
CONNECTING from TF). That helps guarantee that the test is using the
expected subchannel.

The missing ClusterImplLB.requestConnection() doesn't actually matter
much, as cluster manager doesn't propagate connection requests.

Run Details

33624 of 39768 relevant lines covered (84.55%)

0.85 hits per line

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

94.62

/../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 requestConnection() {
    if (childSwitchLb != null) {
      childSwitchLb.requestConnection();
    }
  }

  @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 requestConnection() {
168	if (childSwitchLb != null) {	1✔
169	childSwitchLb.requestConnection();	1✔
170	}
171	}	1✔
172
173	@Override
174	public void shutdown() {
175	if (dropStats != null) {	1✔
176	dropStats.release();	1✔
177	}
178	if (childSwitchLb != null) {	1✔
179	childSwitchLb.shutdown();	1✔
180	if (childLbHelper != null) {	1✔
181	childLbHelper.updateSslContextProviderSupplier(null);	1✔
182	childLbHelper = null;	1✔
183	}
184	}
185	if (xdsClient != null) {	1✔
186	xdsClient = xdsClientPool.returnObject(xdsClient);	1✔
187	}
188	}	1✔
189
190	/**
191	* A decorated {@link LoadBalancer.Helper} that applies configurations for connections
192	* or requests to endpoints in the cluster.
193	*/
194	private final class ClusterImplLbHelper extends ForwardingLoadBalancerHelper {
195	private final AtomicLong inFlights;
196	private ConnectivityState currentState = ConnectivityState.IDLE;	1✔
197	private SubchannelPicker currentPicker = new FixedResultPicker(PickResult.withNoResult());	1✔
198	private List<DropOverload> dropPolicies = Collections.emptyList();	1✔
199	private long maxConcurrentRequests = DEFAULT_PER_CLUSTER_MAX_CONCURRENT_REQUESTS;	1✔
200	@Nullable
201	private SslContextProviderSupplier sslContextProviderSupplier;
202	private Map<String, Struct> filterMetadata = ImmutableMap.of();	1✔
203	@Nullable
204	private final ServerInfo lrsServerInfo;
205
206	private ClusterImplLbHelper(AtomicLong inFlights, @Nullable ServerInfo lrsServerInfo) {	1✔
207	this.inFlights = checkNotNull(inFlights, "inFlights");	1✔
208	this.lrsServerInfo = lrsServerInfo;	1✔
209	}	1✔
210
211	@Override
212	public void updateBalancingState(ConnectivityState newState, SubchannelPicker newPicker) {
213	currentState = newState;	1✔
214	currentPicker = newPicker;	1✔
215	SubchannelPicker picker = new RequestLimitingSubchannelPicker(	1✔
216	newPicker, dropPolicies, maxConcurrentRequests, filterMetadata);
217	delegate().updateBalancingState(newState, picker);	1✔
218	}	1✔
219
220	@Override
221	public Subchannel createSubchannel(CreateSubchannelArgs args) {
222	List<EquivalentAddressGroup> addresses = withAdditionalAttributes(args.getAddresses());	1✔
223	// This value for ClusterLocality is not recommended for general use.
224	// Currently, we extract locality data from the first address, even before the subchannel is
225	// READY.
226	// This is mainly to accommodate scenarios where a Load Balancing API (like "pick first")
227	// might return the subchannel before it is READY. Typically, we wouldn't report load for such
228	// selections because the channel will disregard the chosen (not-ready) subchannel.
229	// However, we needed to ensure this case is handled.
230	ClusterLocality clusterLocality = createClusterLocalityFromAttributes(	1✔
231	args.getAddresses().get(0).getAttributes());	1✔
232	AtomicReference<ClusterLocality> localityAtomicReference = new AtomicReference<>(	1✔
233	clusterLocality);
234	Attributes attrs = args.getAttributes().toBuilder()	1✔
235	.set(ATTR_CLUSTER_LOCALITY, localityAtomicReference)	1✔
236	.build();	1✔
237	args = args.toBuilder().setAddresses(addresses).setAttributes(attrs).build();	1✔
238	final Subchannel subchannel = delegate().createSubchannel(args);	1✔
239
240	return new ForwardingSubchannel() {	1✔
241	@Override
242	public void start(SubchannelStateListener listener) {
243	delegate().start(new SubchannelStateListener() {	1✔
244	@Override
245	public void onSubchannelState(ConnectivityStateInfo newState) {
246	// Do nothing if LB has been shutdown
247	if (xdsClient != null && newState.getState().equals(ConnectivityState.READY)) {	1✔
248	// Get locality based on the connected address attributes
249	ClusterLocality updatedClusterLocality = createClusterLocalityFromAttributes(	1✔
250	subchannel.getConnectedAddressAttributes());	1✔
251	ClusterLocality oldClusterLocality = localityAtomicReference	1✔
252	.getAndSet(updatedClusterLocality);	1✔
253	oldClusterLocality.release();	1✔
254	}
255	listener.onSubchannelState(newState);	1✔
256	}	1✔
257	});
258	}	1✔
259
260	@Override
261	public void shutdown() {
262	localityAtomicReference.get().release();	1✔
263	delegate().shutdown();	1✔
264	}	1✔
265
266	@Override
267	public void updateAddresses(List<EquivalentAddressGroup> addresses) {
268	delegate().updateAddresses(withAdditionalAttributes(addresses));	1✔
269	}	1✔
270
271	@Override
272	protected Subchannel delegate() {
273	return subchannel;	1✔
274	}
275	};
276	}
277
278	private List<EquivalentAddressGroup> withAdditionalAttributes(
279	List<EquivalentAddressGroup> addresses) {
280	List<EquivalentAddressGroup> newAddresses = new ArrayList<>();	1✔
281	for (EquivalentAddressGroup eag : addresses) {	1✔
282	Attributes.Builder attrBuilder = eag.getAttributes().toBuilder().set(	1✔
283	InternalXdsAttributes.ATTR_CLUSTER_NAME, cluster);	1✔
284	if (sslContextProviderSupplier != null) {	1✔
285	attrBuilder.set(	1✔
286	InternalXdsAttributes.ATTR_SSL_CONTEXT_PROVIDER_SUPPLIER,
287	sslContextProviderSupplier);
288	}
289	newAddresses.add(new EquivalentAddressGroup(eag.getAddresses(), attrBuilder.build()));	1✔
290	}	1✔
291	return newAddresses;	1✔
292	}
293
294	private ClusterLocality createClusterLocalityFromAttributes(Attributes addressAttributes) {
295	Locality locality = addressAttributes.get(InternalXdsAttributes.ATTR_LOCALITY);	1✔
296	String localityName = addressAttributes.get(InternalXdsAttributes.ATTR_LOCALITY_NAME);	1✔
297
298	// Endpoint addresses resolved by ClusterResolverLoadBalancer should always contain
299	// attributes with its locality, including endpoints in LOGICAL_DNS clusters.
300	// In case of not (which really shouldn't), loads are aggregated under an empty
301	// locality.
302	if (locality == null) {	1✔
303	locality = Locality.create("", "", "");	×
304	localityName = "";	×
305	}
306
307	final ClusterLocalityStats localityStats =
308	(lrsServerInfo == null)	1✔
309	? null	1✔
310	: xdsClient.addClusterLocalityStats(lrsServerInfo, cluster,	1✔
311	edsServiceName, locality);	1✔
312
313	return new ClusterLocality(localityStats, localityName);	1✔
314	}
315
316	@Override
317	protected Helper delegate() {
318	return helper;	1✔
319	}
320
321	private void updateDropPolicies(List<DropOverload> dropOverloads) {
322	if (!dropPolicies.equals(dropOverloads)) {	1✔
323	dropPolicies = dropOverloads;	1✔
324	updateBalancingState(currentState, currentPicker);	1✔
325	}
326	}	1✔
327
328	private void updateMaxConcurrentRequests(@Nullable Long maxConcurrentRequests) {
329	if (Objects.equals(this.maxConcurrentRequests, maxConcurrentRequests)) {	1✔
330	return;	×
331	}
332	this.maxConcurrentRequests =	1✔
333	maxConcurrentRequests != null	1✔
334	? maxConcurrentRequests	1✔
335	: DEFAULT_PER_CLUSTER_MAX_CONCURRENT_REQUESTS;	1✔
336	updateBalancingState(currentState, currentPicker);	1✔
337	}	1✔
338
339	private void updateSslContextProviderSupplier(@Nullable UpstreamTlsContext tlsContext) {
340	UpstreamTlsContext currentTlsContext =
341	sslContextProviderSupplier != null	1✔
342	? (UpstreamTlsContext)sslContextProviderSupplier.getTlsContext()	1✔
343	: null;	1✔
344	if (Objects.equals(currentTlsContext, tlsContext)) {	1✔
345	return;	1✔
346	}
347	if (sslContextProviderSupplier != null) {	1✔
348	sslContextProviderSupplier.close();	1✔
349	}
350	sslContextProviderSupplier =	1✔
351	tlsContext != null	1✔
352	? new SslContextProviderSupplier(tlsContext,	1✔
353	(TlsContextManager) xdsClient.getSecurityConfig())	1✔
354	: null;	1✔
355	}	1✔
356
357	private void updateFilterMetadata(Map<String, Struct> filterMetadata) {
358	this.filterMetadata = ImmutableMap.copyOf(filterMetadata);	1✔
359	}	1✔
360
361	private class RequestLimitingSubchannelPicker extends SubchannelPicker {
362	private final SubchannelPicker delegate;
363	private final List<DropOverload> dropPolicies;
364	private final long maxConcurrentRequests;
365	private final Map<String, Struct> filterMetadata;
366
367	private RequestLimitingSubchannelPicker(SubchannelPicker delegate,
368	List<DropOverload> dropPolicies, long maxConcurrentRequests,
369	Map<String, Struct> filterMetadata) {	1✔
370	this.delegate = delegate;	1✔
371	this.dropPolicies = dropPolicies;	1✔
372	this.maxConcurrentRequests = maxConcurrentRequests;	1✔
373	this.filterMetadata = checkNotNull(filterMetadata, "filterMetadata");	1✔
374	}	1✔
375
376	@Override
377	public PickResult pickSubchannel(PickSubchannelArgs args) {
378	args.getCallOptions().getOption(ClusterImplLoadBalancerProvider.FILTER_METADATA_CONSUMER)	1✔
379	.accept(filterMetadata);	1✔
380	for (DropOverload dropOverload : dropPolicies) {	1✔
381	int rand = random.nextInt(1_000_000);	1✔
382	if (rand < dropOverload.dropsPerMillion()) {	1✔
383	logger.log(XdsLogLevel.INFO, "Drop request with category: {0}",	1✔
384	dropOverload.category());	1✔
385	if (dropStats != null) {	1✔
386	dropStats.recordDroppedRequest(dropOverload.category());	1✔
387	}
388	return PickResult.withDrop(	1✔
389	Status.UNAVAILABLE.withDescription("Dropped: " + dropOverload.category()));	1✔
390	}
391	}	1✔
392	final PickResult result = delegate.pickSubchannel(args);	1✔
393	if (result.getStatus().isOk() && result.getSubchannel() != null) {	1✔
394	if (enableCircuitBreaking) {	1✔
395	if (inFlights.get() >= maxConcurrentRequests) {	1✔
396	if (dropStats != null) {	1✔
397	dropStats.recordDroppedRequest();	1✔
398	}
399	return PickResult.withDrop(Status.UNAVAILABLE.withDescription(	1✔
400	"Cluster max concurrent requests limit exceeded"));
401	}
402	}
403	final AtomicReference<ClusterLocality> clusterLocality =	1✔
404	result.getSubchannel().getAttributes().get(ATTR_CLUSTER_LOCALITY);	1✔
405
406	if (clusterLocality != null) {	1✔
407	ClusterLocalityStats stats = clusterLocality.get().getClusterLocalityStats();	1✔
408	if (stats != null) {	1✔
409	String localityName =	1✔
410	result.getSubchannel().getAttributes().get(ATTR_CLUSTER_LOCALITY).get()	1✔
411	.getClusterLocalityName();	1✔
412	args.getPickDetailsConsumer().addOptionalLabel("grpc.lb.locality", localityName);	1✔
413
414	ClientStreamTracer.Factory tracerFactory = new CountingStreamTracerFactory(	1✔
415	stats, inFlights, result.getStreamTracerFactory());	1✔
416	ClientStreamTracer.Factory orcaTracerFactory = OrcaPerRequestUtil.getInstance()	1✔
417	.newOrcaClientStreamTracerFactory(tracerFactory, new OrcaPerRpcListener(stats));	1✔
418	return PickResult.withSubchannel(result.getSubchannel(), orcaTracerFactory);	1✔
419	}
420	}
421	}
422	return result;	1✔
423	}
424
425	@Override
426	public String toString() {
427	return MoreObjects.toStringHelper(this).add("delegate", delegate).toString();	×
428	}
429	}
430	}
431
432	private static final class CountingStreamTracerFactory extends
433	ClientStreamTracer.Factory {
434	private final ClusterLocalityStats stats;
435	private final AtomicLong inFlights;
436	@Nullable
437	private final ClientStreamTracer.Factory delegate;
438
439	private CountingStreamTracerFactory(
440	ClusterLocalityStats stats, AtomicLong inFlights,
441	@Nullable ClientStreamTracer.Factory delegate) {	1✔
442	this.stats = checkNotNull(stats, "stats");	1✔
443	this.inFlights = checkNotNull(inFlights, "inFlights");	1✔
444	this.delegate = delegate;	1✔
445	}	1✔
446
447	@Override
448	public ClientStreamTracer newClientStreamTracer(StreamInfo info, Metadata headers) {
449	stats.recordCallStarted();	1✔
450	inFlights.incrementAndGet();	1✔
451	if (delegate == null) {	1✔
452	return new ClientStreamTracer() {	1✔
453	@Override
454	public void streamClosed(Status status) {
455	stats.recordCallFinished(status);	1✔
456	inFlights.decrementAndGet();	1✔
457	}	1✔
458	};
459	}
460	final ClientStreamTracer delegatedTracer = delegate.newClientStreamTracer(info, headers);	×
461	return new ForwardingClientStreamTracer() {	×
462	@Override
463	protected ClientStreamTracer delegate() {
464	return delegatedTracer;	×
465	}
466
467	@Override
468	public void streamClosed(Status status) {
469	stats.recordCallFinished(status);	×
470	inFlights.decrementAndGet();	×
471	delegate().streamClosed(status);	×
472	}	×
473	};
474	}
475	}
476
477	private static final class OrcaPerRpcListener implements OrcaPerRequestReportListener {
478
479	private final ClusterLocalityStats stats;
480
481	private OrcaPerRpcListener(ClusterLocalityStats stats) {	1✔
482	this.stats = checkNotNull(stats, "stats");	1✔
483	}	1✔
484
485	/**
486	* Copies {@link MetricReport#getNamedMetrics()} to {@link ClusterLocalityStats} such that it is
487	* included in the snapshot for the LRS report sent to the LRS server.
488	*/
489	@Override
490	public void onLoadReport(MetricReport report) {
491	stats.recordBackendLoadMetricStats(report.getNamedMetrics());	1✔
492	}	1✔
493	}
494
495	/**
496	* Represents the {@link ClusterLocalityStats} and network locality name of a cluster.
497	*/
498	static final class ClusterLocality {
499	private final ClusterLocalityStats clusterLocalityStats;
500	private final String clusterLocalityName;
501
502	@VisibleForTesting
503	ClusterLocality(ClusterLocalityStats localityStats, String localityName) {	1✔
504	this.clusterLocalityStats = localityStats;	1✔
505	this.clusterLocalityName = localityName;	1✔
506	}	1✔
507
508	ClusterLocalityStats getClusterLocalityStats() {
509	return clusterLocalityStats;	1✔
510	}
511
512	String getClusterLocalityName() {
513	return clusterLocalityName;	1✔
514	}
515
516	@VisibleForTesting
517	void release() {
518	if (clusterLocalityStats != null) {	1✔
519	clusterLocalityStats.release();	1✔
520	}
521	}	1✔
522	}
523	}

grpc / grpc-java / #19472

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