#19249

Committed 24 May 2024 10:08PM UTC coverage: 88.444% (-0.07%) from 88.512%

Build # #19249

Build Type

push

github

Committed by

web-flow

Commit Message

xds: Plumb the Cluster's filterMetadata to RPCs

This will be used by CSM observability, and may get exposed to further
uses in the future.

Run Details

32060 of 36249 relevant lines covered (88.44%)

0.88 hits per line

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

93.97

/../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.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 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<ClusterLocalityStats> ATTR_CLUSTER_LOCALITY_STATS =
      Attributes.Key.create("io.grpc.xds.ClusterImplLoadBalancer.clusterLocalityStats");
  private static final Attributes.Key<String> ATTR_CLUSTER_LOCALITY_NAME =
      Attributes.Key.create("io.grpc.xds.ClusterImplLoadBalancer.clusterLocalityName");

  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.switchTo(config.childPolicy.getProvider());
    childSwitchLb.handleResolvedAddresses(
        resolvedAddresses.toBuilder()
            .setAttributes(attributes)
            .setLoadBalancingPolicyConfig(config.childPolicy.getConfig())
            .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());
      Locality locality = args.getAddresses().get(0).getAttributes().get(
          InternalXdsAttributes.ATTR_LOCALITY);  // all addresses should be in the same locality
      String localityName = args.getAddresses().get(0).getAttributes().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);

      Attributes attrs = args.getAttributes().toBuilder()
          .set(ATTR_CLUSTER_LOCALITY_STATS, localityStats)
          .set(ATTR_CLUSTER_LOCALITY_NAME, localityName)
          .build();
      args = args.toBuilder().setAddresses(addresses).setAttributes(attrs).build();
      final Subchannel subchannel = delegate().createSubchannel(args);

      return new ForwardingSubchannel() {
        @Override
        public void shutdown() {
          if (localityStats != null) {
            localityStats.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;
    }

    @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 ClusterLocalityStats stats =
              result.getSubchannel().getAttributes().get(ATTR_CLUSTER_LOCALITY_STATS);
          if (stats != null) {
            String localityName =
                result.getSubchannel().getAttributes().get(ATTR_CLUSTER_LOCALITY_NAME);
            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());
    }
  }
}

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.EquivalentAddressGroup;
31	import io.grpc.InternalLogId;
32	import io.grpc.LoadBalancer;
33	import io.grpc.Metadata;
34	import io.grpc.Status;
35	import io.grpc.internal.ForwardingClientStreamTracer;
36	import io.grpc.internal.ObjectPool;
37	import io.grpc.services.MetricReport;
38	import io.grpc.util.ForwardingLoadBalancerHelper;
39	import io.grpc.util.ForwardingSubchannel;
40	import io.grpc.util.GracefulSwitchLoadBalancer;
41	import io.grpc.xds.ClusterImplLoadBalancerProvider.ClusterImplConfig;
42	import io.grpc.xds.Endpoints.DropOverload;
43	import io.grpc.xds.EnvoyServerProtoData.UpstreamTlsContext;
44	import io.grpc.xds.ThreadSafeRandom.ThreadSafeRandomImpl;
45	import io.grpc.xds.XdsNameResolverProvider.CallCounterProvider;
46	import io.grpc.xds.client.Bootstrapper.ServerInfo;
47	import io.grpc.xds.client.LoadStatsManager2.ClusterDropStats;
48	import io.grpc.xds.client.LoadStatsManager2.ClusterLocalityStats;
49	import io.grpc.xds.client.Locality;
50	import io.grpc.xds.client.XdsClient;
51	import io.grpc.xds.client.XdsLogger;
52	import io.grpc.xds.client.XdsLogger.XdsLogLevel;
53	import io.grpc.xds.internal.security.SslContextProviderSupplier;
54	import io.grpc.xds.orca.OrcaPerRequestUtil;
55	import io.grpc.xds.orca.OrcaPerRequestUtil.OrcaPerRequestReportListener;
56	import java.util.ArrayList;
57	import java.util.Collections;
58	import java.util.List;
59	import java.util.Map;
60	import java.util.Objects;
61	import java.util.concurrent.atomic.AtomicLong;
62	import javax.annotation.Nullable;
63
64	/**
65	* Load balancer for cluster_impl_experimental LB policy. This LB policy is the child LB policy of
66	* the priority_experimental LB policy and the parent LB policy of the weighted_target_experimental
67	* LB policy in the xDS load balancing hierarchy. This LB policy applies cluster-level
68	* configurations to requests sent to the corresponding cluster, such as drop policies, circuit
69	* breakers.
70	*/
71	final class ClusterImplLoadBalancer extends LoadBalancer {
72
73	@VisibleForTesting
74	static final long DEFAULT_PER_CLUSTER_MAX_CONCURRENT_REQUESTS = 1024L;
75	@VisibleForTesting
76	static boolean enableCircuitBreaking =	1✔
77	Strings.isNullOrEmpty(System.getenv("GRPC_XDS_EXPERIMENTAL_CIRCUIT_BREAKING"))	1✔
78	\|\| Boolean.parseBoolean(System.getenv("GRPC_XDS_EXPERIMENTAL_CIRCUIT_BREAKING"));	1✔
79
80	private static final Attributes.Key<ClusterLocalityStats> ATTR_CLUSTER_LOCALITY_STATS =	1✔
81	Attributes.Key.create("io.grpc.xds.ClusterImplLoadBalancer.clusterLocalityStats");	1✔
82	private static final Attributes.Key<String> ATTR_CLUSTER_LOCALITY_NAME =	1✔
83	Attributes.Key.create("io.grpc.xds.ClusterImplLoadBalancer.clusterLocalityName");	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.switchTo(config.childPolicy.getProvider());	1✔
149	childSwitchLb.handleResolvedAddresses(	1✔
150	resolvedAddresses.toBuilder()	1✔
151	.setAttributes(attributes)	1✔
152	.setLoadBalancingPolicyConfig(config.childPolicy.getConfig())	1✔
153	.build());	1✔
154	return Status.OK;	1✔
155	}
156
157	@Override
158	public void handleNameResolutionError(Status error) {
159	if (childSwitchLb != null) {	1✔
160	childSwitchLb.handleNameResolutionError(error);	1✔
161	} else {
162	helper.updateBalancingState(	1✔
163	ConnectivityState.TRANSIENT_FAILURE, new FixedResultPicker(PickResult.withError(error)));	1✔
164	}
165	}	1✔
166
167	@Override
168	public void shutdown() {
169	if (dropStats != null) {	1✔
170	dropStats.release();	1✔
171	}
172	if (childSwitchLb != null) {	1✔
173	childSwitchLb.shutdown();	1✔
174	if (childLbHelper != null) {	1✔
175	childLbHelper.updateSslContextProviderSupplier(null);	1✔
176	childLbHelper = null;	1✔
177	}
178	}
179	if (xdsClient != null) {	1✔
180	xdsClient = xdsClientPool.returnObject(xdsClient);	1✔
181	}
182	}	1✔
183
184	/**
185	* A decorated {@link LoadBalancer.Helper} that applies configurations for connections
186	* or requests to endpoints in the cluster.
187	*/
188	private final class ClusterImplLbHelper extends ForwardingLoadBalancerHelper {
189	private final AtomicLong inFlights;
190	private ConnectivityState currentState = ConnectivityState.IDLE;	1✔
191	private SubchannelPicker currentPicker = new FixedResultPicker(PickResult.withNoResult());	1✔
192	private List<DropOverload> dropPolicies = Collections.emptyList();	1✔
193	private long maxConcurrentRequests = DEFAULT_PER_CLUSTER_MAX_CONCURRENT_REQUESTS;	1✔
194	@Nullable
195	private SslContextProviderSupplier sslContextProviderSupplier;
196	private Map<String, Struct> filterMetadata = ImmutableMap.of();	1✔
197	@Nullable
198	private final ServerInfo lrsServerInfo;
199
200	private ClusterImplLbHelper(AtomicLong inFlights, @Nullable ServerInfo lrsServerInfo) {	1✔
201	this.inFlights = checkNotNull(inFlights, "inFlights");	1✔
202	this.lrsServerInfo = lrsServerInfo;	1✔
203	}	1✔
204
205	@Override
206	public void updateBalancingState(ConnectivityState newState, SubchannelPicker newPicker) {
207	currentState = newState;	1✔
208	currentPicker = newPicker;	1✔
209	SubchannelPicker picker = new RequestLimitingSubchannelPicker(	1✔
210	newPicker, dropPolicies, maxConcurrentRequests, filterMetadata);
211	delegate().updateBalancingState(newState, picker);	1✔
212	}	1✔
213
214	@Override
215	public Subchannel createSubchannel(CreateSubchannelArgs args) {
216	List<EquivalentAddressGroup> addresses = withAdditionalAttributes(args.getAddresses());	1✔
217	Locality locality = args.getAddresses().get(0).getAttributes().get(	1✔
218	InternalXdsAttributes.ATTR_LOCALITY); // all addresses should be in the same locality
219	String localityName = args.getAddresses().get(0).getAttributes().get(	1✔
220	InternalXdsAttributes.ATTR_LOCALITY_NAME);
221	// Endpoint addresses resolved by ClusterResolverLoadBalancer should always contain
222	// attributes with its locality, including endpoints in LOGICAL_DNS clusters.
223	// In case of not (which really shouldn't), loads are aggregated under an empty locality.
224	if (locality == null) {	1✔
225	locality = Locality.create("", "", "");	×
226	localityName = "";	×
227	}
228	final ClusterLocalityStats localityStats =
229	(lrsServerInfo == null)	1✔
230	? null	1✔
231	: xdsClient.addClusterLocalityStats(lrsServerInfo, cluster,	1✔
232	edsServiceName, locality);	1✔
233
234	Attributes attrs = args.getAttributes().toBuilder()	1✔
235	.set(ATTR_CLUSTER_LOCALITY_STATS, localityStats)	1✔
236	.set(ATTR_CLUSTER_LOCALITY_NAME, localityName)	1✔
237	.build();	1✔
238	args = args.toBuilder().setAddresses(addresses).setAttributes(attrs).build();	1✔
239	final Subchannel subchannel = delegate().createSubchannel(args);	1✔
240
241	return new ForwardingSubchannel() {	1✔
242	@Override
243	public void shutdown() {
244	if (localityStats != null) {	1✔
245	localityStats.release();	1✔
246	}
247	delegate().shutdown();	1✔
248	}	1✔
249
250	@Override
251	public void updateAddresses(List<EquivalentAddressGroup> addresses) {
252	delegate().updateAddresses(withAdditionalAttributes(addresses));	1✔
253	}	1✔
254
255	@Override
256	protected Subchannel delegate() {
257	return subchannel;	1✔
258	}
259	};
260	}
261
262	private List<EquivalentAddressGroup> withAdditionalAttributes(
263	List<EquivalentAddressGroup> addresses) {
264	List<EquivalentAddressGroup> newAddresses = new ArrayList<>();	1✔
265	for (EquivalentAddressGroup eag : addresses) {	1✔
266	Attributes.Builder attrBuilder = eag.getAttributes().toBuilder().set(	1✔
267	InternalXdsAttributes.ATTR_CLUSTER_NAME, cluster);	1✔
268	if (sslContextProviderSupplier != null) {	1✔
269	attrBuilder.set(	1✔
270	InternalXdsAttributes.ATTR_SSL_CONTEXT_PROVIDER_SUPPLIER,
271	sslContextProviderSupplier);
272	}
273	newAddresses.add(new EquivalentAddressGroup(eag.getAddresses(), attrBuilder.build()));	1✔
274	}	1✔
275	return newAddresses;	1✔
276	}
277
278	@Override
279	protected Helper delegate() {
280	return helper;	1✔
281	}
282
283	private void updateDropPolicies(List<DropOverload> dropOverloads) {
284	if (!dropPolicies.equals(dropOverloads)) {	1✔
285	dropPolicies = dropOverloads;	1✔
286	updateBalancingState(currentState, currentPicker);	1✔
287	}
288	}	1✔
289
290	private void updateMaxConcurrentRequests(@Nullable Long maxConcurrentRequests) {
291	if (Objects.equals(this.maxConcurrentRequests, maxConcurrentRequests)) {	1✔
292	return;	×
293	}
294	this.maxConcurrentRequests =	1✔
295	maxConcurrentRequests != null	1✔
296	? maxConcurrentRequests	1✔
297	: DEFAULT_PER_CLUSTER_MAX_CONCURRENT_REQUESTS;	1✔
298	updateBalancingState(currentState, currentPicker);	1✔
299	}	1✔
300
301	private void updateSslContextProviderSupplier(@Nullable UpstreamTlsContext tlsContext) {
302	UpstreamTlsContext currentTlsContext =
303	sslContextProviderSupplier != null	1✔
304	? (UpstreamTlsContext)sslContextProviderSupplier.getTlsContext()	1✔
305	: null;	1✔
306	if (Objects.equals(currentTlsContext, tlsContext)) {	1✔
307	return;	1✔
308	}
309	if (sslContextProviderSupplier != null) {	1✔
310	sslContextProviderSupplier.close();	1✔
311	}
312	sslContextProviderSupplier =	1✔
313	tlsContext != null	1✔
314	? new SslContextProviderSupplier(tlsContext,	1✔
315	(TlsContextManager) xdsClient.getSecurityConfig())	1✔
316	: null;	1✔
317	}	1✔
318
319	private void updateFilterMetadata(Map<String, Struct> filterMetadata) {
320	this.filterMetadata = ImmutableMap.copyOf(filterMetadata);	1✔
321	}	1✔
322
323	private class RequestLimitingSubchannelPicker extends SubchannelPicker {
324	private final SubchannelPicker delegate;
325	private final List<DropOverload> dropPolicies;
326	private final long maxConcurrentRequests;
327	private final Map<String, Struct> filterMetadata;
328
329	private RequestLimitingSubchannelPicker(SubchannelPicker delegate,
330	List<DropOverload> dropPolicies, long maxConcurrentRequests,
331	Map<String, Struct> filterMetadata) {	1✔
332	this.delegate = delegate;	1✔
333	this.dropPolicies = dropPolicies;	1✔
334	this.maxConcurrentRequests = maxConcurrentRequests;	1✔
335	this.filterMetadata = checkNotNull(filterMetadata, "filterMetadata");	1✔
336	}	1✔
337
338	@Override
339	public PickResult pickSubchannel(PickSubchannelArgs args) {
340	args.getCallOptions().getOption(ClusterImplLoadBalancerProvider.FILTER_METADATA_CONSUMER)	1✔
341	.accept(filterMetadata);	1✔
342	for (DropOverload dropOverload : dropPolicies) {	1✔
343	int rand = random.nextInt(1_000_000);	1✔
344	if (rand < dropOverload.dropsPerMillion()) {	1✔
345	logger.log(XdsLogLevel.INFO, "Drop request with category: {0}",	1✔
346	dropOverload.category());	1✔
347	if (dropStats != null) {	1✔
348	dropStats.recordDroppedRequest(dropOverload.category());	1✔
349	}
350	return PickResult.withDrop(	1✔
351	Status.UNAVAILABLE.withDescription("Dropped: " + dropOverload.category()));	1✔
352	}
353	}	1✔
354	final PickResult result = delegate.pickSubchannel(args);	1✔
355	if (result.getStatus().isOk() && result.getSubchannel() != null) {	1✔
356	if (enableCircuitBreaking) {	1✔
357	if (inFlights.get() >= maxConcurrentRequests) {	1✔
358	if (dropStats != null) {	1✔
359	dropStats.recordDroppedRequest();	1✔
360	}
361	return PickResult.withDrop(Status.UNAVAILABLE.withDescription(	1✔
362	"Cluster max concurrent requests limit exceeded"));
363	}
364	}
365	final ClusterLocalityStats stats =	1✔
366	result.getSubchannel().getAttributes().get(ATTR_CLUSTER_LOCALITY_STATS);	1✔
367	if (stats != null) {	1✔
368	String localityName =	1✔
369	result.getSubchannel().getAttributes().get(ATTR_CLUSTER_LOCALITY_NAME);	1✔
370	args.getPickDetailsConsumer().addOptionalLabel("grpc.lb.locality", localityName);	1✔
371
372	ClientStreamTracer.Factory tracerFactory = new CountingStreamTracerFactory(	1✔
373	stats, inFlights, result.getStreamTracerFactory());	1✔
374	ClientStreamTracer.Factory orcaTracerFactory = OrcaPerRequestUtil.getInstance()	1✔
375	.newOrcaClientStreamTracerFactory(tracerFactory, new OrcaPerRpcListener(stats));	1✔
376	return PickResult.withSubchannel(result.getSubchannel(), orcaTracerFactory);	1✔
377	}
378	}
379	return result;	1✔
380	}
381
382	@Override
383	public String toString() {
384	return MoreObjects.toStringHelper(this).add("delegate", delegate).toString();	×
385	}
386	}
387	}
388
389	private static final class CountingStreamTracerFactory extends
390	ClientStreamTracer.Factory {
391	private final ClusterLocalityStats stats;
392	private final AtomicLong inFlights;
393	@Nullable
394	private final ClientStreamTracer.Factory delegate;
395
396	private CountingStreamTracerFactory(
397	ClusterLocalityStats stats, AtomicLong inFlights,
398	@Nullable ClientStreamTracer.Factory delegate) {	1✔
399	this.stats = checkNotNull(stats, "stats");	1✔
400	this.inFlights = checkNotNull(inFlights, "inFlights");	1✔
401	this.delegate = delegate;	1✔
402	}	1✔
403
404	@Override
405	public ClientStreamTracer newClientStreamTracer(StreamInfo info, Metadata headers) {
406	stats.recordCallStarted();	1✔
407	inFlights.incrementAndGet();	1✔
408	if (delegate == null) {	1✔
409	return new ClientStreamTracer() {	1✔
410	@Override
411	public void streamClosed(Status status) {
412	stats.recordCallFinished(status);	1✔
413	inFlights.decrementAndGet();	1✔
414	}	1✔
415	};
416	}
417	final ClientStreamTracer delegatedTracer = delegate.newClientStreamTracer(info, headers);	×
418	return new ForwardingClientStreamTracer() {	×
419	@Override
420	protected ClientStreamTracer delegate() {
421	return delegatedTracer;	×
422	}
423
424	@Override
425	public void streamClosed(Status status) {
426	stats.recordCallFinished(status);	×
427	inFlights.decrementAndGet();	×
428	delegate().streamClosed(status);	×
429	}	×
430	};
431	}
432	}
433
434	private static final class OrcaPerRpcListener implements OrcaPerRequestReportListener {
435
436	private final ClusterLocalityStats stats;
437
438	private OrcaPerRpcListener(ClusterLocalityStats stats) {	1✔
439	this.stats = checkNotNull(stats, "stats");	1✔
440	}	1✔
441
442	/**
443	* Copies {@link MetricReport#getNamedMetrics()} to {@link ClusterLocalityStats} such that it is
444	* included in the snapshot for the LRS report sent to the LRS server.
445	*/
446	@Override
447	public void onLoadReport(MetricReport report) {
448	stats.recordBackendLoadMetricStats(report.getNamedMetrics());	1✔
449	}	1✔
450	}
451	}

grpc / grpc-java / #19249

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