#19447

Committed 31 Aug 2024 11:07PM CUT coverage: 84.561% (+0.04%) from 84.523%

Build # #19447

Build Type

push

github

Committed by

web-flow

Commit Message

xds: Fix load reporting when pick first is used for locality-routing. (#11495)

* Determine subchannel's network locality from connected address, instead of assuming that all addresses for a subchannel are in the same locality.

Run Details

33554 of 39680 relevant lines covered (84.56%)

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) {
              if (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	if (newState.getState().equals(ConnectivityState.READY)) {	1✔
240	// Get locality based on the connected address attributes
241	ClusterLocality updatedClusterLocality = createClusterLocalityFromAttributes(	1✔
242	subchannel.getConnectedAddressAttributes());	1✔
243	ClusterLocality oldClusterLocality = localityAtomicReference	1✔
244	.getAndSet(updatedClusterLocality);	1✔
245	oldClusterLocality.release();	1✔
246	}
247	listener.onSubchannelState(newState);	1✔
248	}	1✔
249	});
250	}	1✔
251
252	@Override
253	public void shutdown() {
254	localityAtomicReference.get().release();	1✔
255	delegate().shutdown();	1✔
256	}	1✔
257
258	@Override
259	public void updateAddresses(List<EquivalentAddressGroup> addresses) {
260	delegate().updateAddresses(withAdditionalAttributes(addresses));	1✔
261	}	1✔
262
263	@Override
264	protected Subchannel delegate() {
265	return subchannel;	1✔
266	}
267	};
268	}
269
270	private List<EquivalentAddressGroup> withAdditionalAttributes(
271	List<EquivalentAddressGroup> addresses) {
272	List<EquivalentAddressGroup> newAddresses = new ArrayList<>();	1✔
273	for (EquivalentAddressGroup eag : addresses) {	1✔
274	Attributes.Builder attrBuilder = eag.getAttributes().toBuilder().set(	1✔
275	InternalXdsAttributes.ATTR_CLUSTER_NAME, cluster);	1✔
276	if (sslContextProviderSupplier != null) {	1✔
277	attrBuilder.set(	1✔
278	InternalXdsAttributes.ATTR_SSL_CONTEXT_PROVIDER_SUPPLIER,
279	sslContextProviderSupplier);
280	}
281	newAddresses.add(new EquivalentAddressGroup(eag.getAddresses(), attrBuilder.build()));	1✔
282	}	1✔
283	return newAddresses;	1✔
284	}
285
286	private ClusterLocality createClusterLocalityFromAttributes(Attributes addressAttributes) {
287	Locality locality = addressAttributes.get(InternalXdsAttributes.ATTR_LOCALITY);	1✔
288	String localityName = addressAttributes.get(InternalXdsAttributes.ATTR_LOCALITY_NAME);	1✔
289
290	// Endpoint addresses resolved by ClusterResolverLoadBalancer should always contain
291	// attributes with its locality, including endpoints in LOGICAL_DNS clusters.
292	// In case of not (which really shouldn't), loads are aggregated under an empty
293	// locality.
294	if (locality == null) {	1✔
295	locality = Locality.create("", "", "");	×
296	localityName = "";	×
297	}
298
299	final ClusterLocalityStats localityStats =
300	(lrsServerInfo == null)	1✔
301	? null	1✔
302	: xdsClient.addClusterLocalityStats(lrsServerInfo, cluster,	1✔
303	edsServiceName, locality);	1✔
304
305	return new ClusterLocality(localityStats, localityName);	1✔
306	}
307
308	@Override
309	protected Helper delegate() {
310	return helper;	1✔
311	}
312
313	private void updateDropPolicies(List<DropOverload> dropOverloads) {
314	if (!dropPolicies.equals(dropOverloads)) {	1✔
315	dropPolicies = dropOverloads;	1✔
316	updateBalancingState(currentState, currentPicker);	1✔
317	}
318	}	1✔
319
320	private void updateMaxConcurrentRequests(@Nullable Long maxConcurrentRequests) {
321	if (Objects.equals(this.maxConcurrentRequests, maxConcurrentRequests)) {	1✔
322	return;	×
323	}
324	this.maxConcurrentRequests =	1✔
325	maxConcurrentRequests != null	1✔
326	? maxConcurrentRequests	1✔
327	: DEFAULT_PER_CLUSTER_MAX_CONCURRENT_REQUESTS;	1✔
328	updateBalancingState(currentState, currentPicker);	1✔
329	}	1✔
330
331	private void updateSslContextProviderSupplier(@Nullable UpstreamTlsContext tlsContext) {
332	UpstreamTlsContext currentTlsContext =
333	sslContextProviderSupplier != null	1✔
334	? (UpstreamTlsContext)sslContextProviderSupplier.getTlsContext()	1✔
335	: null;	1✔
336	if (Objects.equals(currentTlsContext, tlsContext)) {	1✔
337	return;	1✔
338	}
339	if (sslContextProviderSupplier != null) {	1✔
340	sslContextProviderSupplier.close();	1✔
341	}
342	sslContextProviderSupplier =	1✔
343	tlsContext != null	1✔
344	? new SslContextProviderSupplier(tlsContext,	1✔
345	(TlsContextManager) xdsClient.getSecurityConfig())	1✔
346	: null;	1✔
347	}	1✔
348
349	private void updateFilterMetadata(Map<String, Struct> filterMetadata) {
350	this.filterMetadata = ImmutableMap.copyOf(filterMetadata);	1✔
351	}	1✔
352
353	private class RequestLimitingSubchannelPicker extends SubchannelPicker {
354	private final SubchannelPicker delegate;
355	private final List<DropOverload> dropPolicies;
356	private final long maxConcurrentRequests;
357	private final Map<String, Struct> filterMetadata;
358
359	private RequestLimitingSubchannelPicker(SubchannelPicker delegate,
360	List<DropOverload> dropPolicies, long maxConcurrentRequests,
361	Map<String, Struct> filterMetadata) {	1✔
362	this.delegate = delegate;	1✔
363	this.dropPolicies = dropPolicies;	1✔
364	this.maxConcurrentRequests = maxConcurrentRequests;	1✔
365	this.filterMetadata = checkNotNull(filterMetadata, "filterMetadata");	1✔
366	}	1✔
367
368	@Override
369	public PickResult pickSubchannel(PickSubchannelArgs args) {
370	args.getCallOptions().getOption(ClusterImplLoadBalancerProvider.FILTER_METADATA_CONSUMER)	1✔
371	.accept(filterMetadata);	1✔
372	for (DropOverload dropOverload : dropPolicies) {	1✔
373	int rand = random.nextInt(1_000_000);	1✔
374	if (rand < dropOverload.dropsPerMillion()) {	1✔
375	logger.log(XdsLogLevel.INFO, "Drop request with category: {0}",	1✔
376	dropOverload.category());	1✔
377	if (dropStats != null) {	1✔
378	dropStats.recordDroppedRequest(dropOverload.category());	1✔
379	}
380	return PickResult.withDrop(	1✔
381	Status.UNAVAILABLE.withDescription("Dropped: " + dropOverload.category()));	1✔
382	}
383	}	1✔
384	final PickResult result = delegate.pickSubchannel(args);	1✔
385	if (result.getStatus().isOk() && result.getSubchannel() != null) {	1✔
386	if (enableCircuitBreaking) {	1✔
387	if (inFlights.get() >= maxConcurrentRequests) {	1✔
388	if (dropStats != null) {	1✔
389	dropStats.recordDroppedRequest();	1✔
390	}
391	return PickResult.withDrop(Status.UNAVAILABLE.withDescription(	1✔
392	"Cluster max concurrent requests limit exceeded"));
393	}
394	}
395	final AtomicReference<ClusterLocality> clusterLocality =	1✔
396	result.getSubchannel().getAttributes().get(ATTR_CLUSTER_LOCALITY);	1✔
397
398	if (clusterLocality != null) {	1✔
399	ClusterLocalityStats stats = clusterLocality.get().getClusterLocalityStats();	1✔
400	if (stats != null) {	1✔
401	String localityName =	1✔
402	result.getSubchannel().getAttributes().get(ATTR_CLUSTER_LOCALITY).get()	1✔
403	.getClusterLocalityName();	1✔
404	args.getPickDetailsConsumer().addOptionalLabel("grpc.lb.locality", localityName);	1✔
405
406	ClientStreamTracer.Factory tracerFactory = new CountingStreamTracerFactory(	1✔
407	stats, inFlights, result.getStreamTracerFactory());	1✔
408	ClientStreamTracer.Factory orcaTracerFactory = OrcaPerRequestUtil.getInstance()	1✔
409	.newOrcaClientStreamTracerFactory(tracerFactory, new OrcaPerRpcListener(stats));	1✔
410	return PickResult.withSubchannel(result.getSubchannel(), orcaTracerFactory);	1✔
411	}
412	}
413	}
414	return result;	1✔
415	}
416
417	@Override
418	public String toString() {
419	return MoreObjects.toStringHelper(this).add("delegate", delegate).toString();	×
420	}
421	}
422	}
423
424	private static final class CountingStreamTracerFactory extends
425	ClientStreamTracer.Factory {
426	private final ClusterLocalityStats stats;
427	private final AtomicLong inFlights;
428	@Nullable
429	private final ClientStreamTracer.Factory delegate;
430
431	private CountingStreamTracerFactory(
432	ClusterLocalityStats stats, AtomicLong inFlights,
433	@Nullable ClientStreamTracer.Factory delegate) {	1✔
434	this.stats = checkNotNull(stats, "stats");	1✔
435	this.inFlights = checkNotNull(inFlights, "inFlights");	1✔
436	this.delegate = delegate;	1✔
437	}	1✔
438
439	@Override
440	public ClientStreamTracer newClientStreamTracer(StreamInfo info, Metadata headers) {
441	stats.recordCallStarted();	1✔
442	inFlights.incrementAndGet();	1✔
443	if (delegate == null) {	1✔
444	return new ClientStreamTracer() {	1✔
445	@Override
446	public void streamClosed(Status status) {
447	stats.recordCallFinished(status);	1✔
448	inFlights.decrementAndGet();	1✔
449	}	1✔
450	};
451	}
452	final ClientStreamTracer delegatedTracer = delegate.newClientStreamTracer(info, headers);	×
453	return new ForwardingClientStreamTracer() {	×
454	@Override
455	protected ClientStreamTracer delegate() {
456	return delegatedTracer;	×
457	}
458
459	@Override
460	public void streamClosed(Status status) {
461	stats.recordCallFinished(status);	×
462	inFlights.decrementAndGet();	×
463	delegate().streamClosed(status);	×
464	}	×
465	};
466	}
467	}
468
469	private static final class OrcaPerRpcListener implements OrcaPerRequestReportListener {
470
471	private final ClusterLocalityStats stats;
472
473	private OrcaPerRpcListener(ClusterLocalityStats stats) {	1✔
474	this.stats = checkNotNull(stats, "stats");	1✔
475	}	1✔
476
477	/**
478	* Copies {@link MetricReport#getNamedMetrics()} to {@link ClusterLocalityStats} such that it is
479	* included in the snapshot for the LRS report sent to the LRS server.
480	*/
481	@Override
482	public void onLoadReport(MetricReport report) {
483	stats.recordBackendLoadMetricStats(report.getNamedMetrics());	1✔
484	}	1✔
485	}
486
487	/**
488	* Represents the {@link ClusterLocalityStats} and network locality name of a cluster.
489	*/
490	static final class ClusterLocality {
491	private final ClusterLocalityStats clusterLocalityStats;
492	private final String clusterLocalityName;
493
494	@VisibleForTesting
495	ClusterLocality(ClusterLocalityStats localityStats, String localityName) {	1✔
496	this.clusterLocalityStats = localityStats;	1✔
497	this.clusterLocalityName = localityName;	1✔
498	}	1✔
499
500	ClusterLocalityStats getClusterLocalityStats() {
501	return clusterLocalityStats;	1✔
502	}
503
504	String getClusterLocalityName() {
505	return clusterLocalityName;	1✔
506	}
507
508	@VisibleForTesting
509	void release() {
510	if (clusterLocalityStats != null) {	1✔
511	clusterLocalityStats.release();	1✔
512	}
513	}	1✔
514	}
515	}

grpc / grpc-java / #19447

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