#19167

Committed 22 Apr 2024 02:48PM UTC coverage: 88.084% (-0.01%) from 88.096%

Build # #19167

Build Type

push

github

Committed by

ejona86

Commit Message

util: Remove deactivation and GracefulSwitchLb from MultiChildLb

It is easy to manage these things outside of MultiChildLb and it makes
the shared code easier and use less memory. In particular, we don't want
to use many instances of GracefulSwitchLb in virtually every policy
simply because it was needed in one or two cases.

Run Details

31195 of 35415 relevant lines covered (88.08%)

0.88 hits per line

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

94.05

/../util/src/main/java/io/grpc/util/MultiChildLoadBalancer.java

/*
 * Copyright 2023 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.util;

import static com.google.common.base.Preconditions.checkArgument;
import static com.google.common.base.Preconditions.checkNotNull;
import static io.grpc.ConnectivityState.CONNECTING;
import static io.grpc.ConnectivityState.IDLE;
import static io.grpc.ConnectivityState.READY;
import static io.grpc.ConnectivityState.SHUTDOWN;
import static io.grpc.ConnectivityState.TRANSIENT_FAILURE;

import com.google.common.annotations.VisibleForTesting;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableMap;
import io.grpc.Attributes;
import io.grpc.ConnectivityState;
import io.grpc.EquivalentAddressGroup;
import io.grpc.Internal;
import io.grpc.LoadBalancer;
import io.grpc.LoadBalancerProvider;
import io.grpc.Status;
import io.grpc.internal.PickFirstLoadBalancerProvider;
import java.net.SocketAddress;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.logging.Level;
import java.util.logging.Logger;
import javax.annotation.Nullable;

/**
 * A base load balancing policy for those policies which has multiple children such as
 * ClusterManager or the petiole policies.  For internal use only.
 */
@Internal
public abstract class MultiChildLoadBalancer extends LoadBalancer {

  private static final Logger logger = Logger.getLogger(MultiChildLoadBalancer.class.getName());
  private final Map<Object, ChildLbState> childLbStates = new LinkedHashMap<>();
  private final Helper helper;
  // Set to true if currently in the process of handling resolved addresses.
  protected boolean resolvingAddresses;

  protected final LoadBalancerProvider pickFirstLbProvider = new PickFirstLoadBalancerProvider();

  protected ConnectivityState currentConnectivityState;


  protected MultiChildLoadBalancer(Helper helper) {
    this.helper = checkNotNull(helper, "helper");
    logger.log(Level.FINE, "Created");
  }

  /**
   * Using the state of all children will calculate the current connectivity state,
   * update fields, generate a picker and then call
   * {@link Helper#updateBalancingState(ConnectivityState, SubchannelPicker)}.
   */
  protected abstract void updateOverallBalancingState();

  /**
   * Override to utilize parsing of the policy configuration or alternative helper/lb generation.
   */
  protected Map<Object, ChildLbState> createChildLbMap(ResolvedAddresses resolvedAddresses) {
    Map<Object, ChildLbState> childLbMap = new HashMap<>();
    List<EquivalentAddressGroup> addresses = resolvedAddresses.getAddresses();
    for (EquivalentAddressGroup eag : addresses) {
      Endpoint endpoint = new Endpoint(eag); // keys need to be just addresses
      ChildLbState existingChildLbState = childLbStates.get(endpoint);
      if (existingChildLbState != null) {
        childLbMap.put(endpoint, existingChildLbState);
      } else {
        childLbMap.put(endpoint,
            createChildLbState(endpoint, null, getInitialPicker(), resolvedAddresses));
      }
    }
    return childLbMap;
  }

  /**
   * Override to create an instance of a subclass.
   */
  protected ChildLbState createChildLbState(Object key, Object policyConfig,
      SubchannelPicker initialPicker, ResolvedAddresses resolvedAddresses) {
    return new ChildLbState(key, pickFirstLbProvider, policyConfig, initialPicker);
  }

  /**
   *   Override to completely replace the default logic or to do additional activities.
   */
  @Override
  public Status acceptResolvedAddresses(ResolvedAddresses resolvedAddresses) {
    try {
      resolvingAddresses = true;

      // process resolvedAddresses to update children
      AcceptResolvedAddrRetVal acceptRetVal = acceptResolvedAddressesInternal(resolvedAddresses);
      if (!acceptRetVal.status.isOk()) {
        return acceptRetVal.status;
      }

      // Update the picker and our connectivity state
      updateOverallBalancingState();

      // shutdown removed children
      shutdownRemoved(acceptRetVal.removedChildren);
      return acceptRetVal.status;
    } finally {
      resolvingAddresses = false;
    }
  }

  /**
   * Override this if your keys are not of type Endpoint.
   * @param key Key to identify the ChildLbState
   * @param resolvedAddresses list of addresses which include attributes
   * @param childConfig a load balancing policy config. This field is optional.
   * @return a fully loaded ResolvedAddresses object for the specified key
   */
  protected ResolvedAddresses getChildAddresses(Object key, ResolvedAddresses resolvedAddresses,
      Object childConfig) {
    Endpoint endpointKey;
    if (key instanceof EquivalentAddressGroup) {
      endpointKey = new Endpoint((EquivalentAddressGroup) key);
    } else {
      checkArgument(key instanceof Endpoint, "key is wrong type");
      endpointKey = (Endpoint) key;
    }

    // Retrieve the non-stripped version
    EquivalentAddressGroup eagToUse = null;
    for (EquivalentAddressGroup currEag : resolvedAddresses.getAddresses()) {
      if (endpointKey.equals(new Endpoint(currEag))) {
        eagToUse = currEag;
        break;
      }
    }

    checkNotNull(eagToUse, key + " no longer present in load balancer children");

    return resolvedAddresses.toBuilder()
        .setAddresses(Collections.singletonList(eagToUse))
        .setAttributes(Attributes.newBuilder().set(IS_PETIOLE_POLICY, true).build())
        .setLoadBalancingPolicyConfig(childConfig)
        .build();
  }

  /**
   * Handle the name resolution error.
   *
   * <p/>Override if you need special handling.
   */
  @Override
  public void handleNameResolutionError(Status error) {
    if (currentConnectivityState != READY)  {
      helper.updateBalancingState(TRANSIENT_FAILURE, getErrorPicker(error));
    }
  }

  /**
   * Handle the name resolution error only for the specified child.
   *
   * <p/>Override if you need special handling.
   */
  protected void handleNameResolutionError(ChildLbState child, Status error) {
    child.lb.handleNameResolutionError(error);
  }

  /**
   * Creates a picker representing the state before any connections have been established.
   *
   * <p/>Override to produce a custom picker.
   */
  protected SubchannelPicker getInitialPicker() {
    return new FixedResultPicker(PickResult.withNoResult());
  }

  /**
   * Creates a new picker representing an error status.
   *
   * <p/>Override to produce a custom picker when there are errors.
   */
  protected SubchannelPicker getErrorPicker(Status error)  {
    return new FixedResultPicker(PickResult.withError(error));
  }

  @Override
  public void shutdown() {
    logger.log(Level.FINE, "Shutdown");
    for (ChildLbState state : childLbStates.values()) {
      state.shutdown();
    }
    childLbStates.clear();
  }

  /**
   *   This does the work to update the child map and calculate which children have been removed.
   *   You must call {@link #updateOverallBalancingState} to update the picker
   *   and call {@link #shutdownRemoved(List)} to shutdown the endpoints that have been removed.
    */
  protected final AcceptResolvedAddrRetVal acceptResolvedAddressesInternal(
      ResolvedAddresses resolvedAddresses) {
    logger.log(Level.FINE, "Received resolution result: {0}", resolvedAddresses);

    // Subclass handles any special manipulation to create appropriate types of keyed ChildLbStates
    Map<Object, ChildLbState> newChildren = createChildLbMap(resolvedAddresses);

    // Handle error case
    if (newChildren.isEmpty()) {
      Status unavailableStatus = Status.UNAVAILABLE.withDescription(
          "NameResolver returned no usable address. " + resolvedAddresses);
      handleNameResolutionError(unavailableStatus);
      return new AcceptResolvedAddrRetVal(unavailableStatus, null);
    }

    addMissingChildren(newChildren);

    updateChildrenWithResolvedAddresses(resolvedAddresses, newChildren);

    return new AcceptResolvedAddrRetVal(Status.OK, getRemovedChildren(newChildren.keySet()));
  }

  protected final void addMissingChildren(Map<Object, ChildLbState> newChildren) {
    // Do adds and identify reused children
    for (Map.Entry<Object, ChildLbState> entry : newChildren.entrySet()) {
      final Object key = entry.getKey();
      if (!childLbStates.containsKey(key)) {
        childLbStates.put(key, entry.getValue());
      }
    }
  }

  protected final void updateChildrenWithResolvedAddresses(ResolvedAddresses resolvedAddresses,
                                                     Map<Object, ChildLbState> newChildren) {
    for (Map.Entry<Object, ChildLbState> entry : newChildren.entrySet()) {
      Object childConfig = entry.getValue().getConfig();
      ChildLbState childLbState = childLbStates.get(entry.getKey());
      ResolvedAddresses childAddresses =
          getChildAddresses(entry.getKey(), resolvedAddresses, childConfig);
      childLbState.setResolvedAddresses(childAddresses); // update child
      childLbState.lb.handleResolvedAddresses(childAddresses); // update child LB
    }
  }

  /**
   * Identifies which children have been removed (are not part of the newChildKeys).
   */
  protected final List<ChildLbState> getRemovedChildren(Set<Object> newChildKeys) {
    List<ChildLbState> removedChildren = new ArrayList<>();
    // Do removals
    for (Object key : ImmutableList.copyOf(childLbStates.keySet())) {
      if (!newChildKeys.contains(key)) {
        ChildLbState childLbState = childLbStates.remove(key);
        removedChildren.add(childLbState);
      }
    }
    return removedChildren;
  }

  protected final void shutdownRemoved(List<ChildLbState> removedChildren) {
    // Do shutdowns after updating picker to reduce the chance of failing an RPC by picking a
    // subchannel that has been shutdown.
    for (ChildLbState childLbState : removedChildren) {
      childLbState.shutdown();
    }
  }

  @Nullable
  protected static ConnectivityState aggregateState(
      @Nullable ConnectivityState overallState, ConnectivityState childState) {
    if (overallState == null) {
      return childState;
    }
    if (overallState == READY || childState == READY) {
      return READY;
    }
    if (overallState == CONNECTING || childState == CONNECTING) {
      return CONNECTING;
    }
    if (overallState == IDLE || childState == IDLE) {
      return IDLE;
    }
    return overallState;
  }

  protected final Helper getHelper() {
    return helper;
  }

  @VisibleForTesting
  public final ImmutableMap<Object, ChildLbState> getImmutableChildMap() {
    return ImmutableMap.copyOf(childLbStates);
  }

  @VisibleForTesting
  public final Collection<ChildLbState> getChildLbStates() {
    return childLbStates.values();
  }

  @VisibleForTesting
  public final ChildLbState getChildLbState(Object key) {
    if (key == null) {
      return null;
    }
    if (key instanceof EquivalentAddressGroup) {
      key = new Endpoint((EquivalentAddressGroup) key);
    }
    return childLbStates.get(key);
  }

  @VisibleForTesting
  public final ChildLbState getChildLbStateEag(EquivalentAddressGroup eag) {
    return getChildLbState(new Endpoint(eag));
  }

  /**
   * Filters out non-ready child load balancers (subchannels).
   */
  protected final List<ChildLbState> getReadyChildren() {
    List<ChildLbState> activeChildren = new ArrayList<>();
    for (ChildLbState child : getChildLbStates()) {
      if (child.getCurrentState() == READY) {
        activeChildren.add(child);
      }
    }
    return activeChildren;
  }

  /**
   * This represents the state of load balancer children.  Each endpoint (represented by an
   * EquivalentAddressGroup or EDS string) will have a separate ChildLbState which in turn will
   * have a single child LoadBalancer created from the provided factory.
   *
   * <p>A ChildLbStateHelper is the glue between ChildLbState and the helpers associated with the
   * petiole policy above and the PickFirstLoadBalancer's helper below.
   *
   * <p>If you wish to store additional state information related to each subchannel, then extend
   * this class.
   */
  public class ChildLbState {
    private final Object key;
    private ResolvedAddresses resolvedAddresses;
    private final Object config;

    private final LoadBalancer lb;
    private ConnectivityState currentState;
    private SubchannelPicker currentPicker;

    public ChildLbState(Object key, LoadBalancer.Factory policyFactory, Object childConfig,
          SubchannelPicker initialPicker) {
      this.key = key;
      this.currentPicker = initialPicker;
      this.config = childConfig;
      this.lb = policyFactory.newLoadBalancer(createChildHelper());
      this.currentState = CONNECTING;
    }

    protected ChildLbStateHelper createChildHelper() {
      return new ChildLbStateHelper();
    }

    /**
     * Override for unique behavior such as delayed shutdowns of subchannels.
     */
    protected void shutdown() {
      lb.shutdown();
      this.currentState = SHUTDOWN;
      logger.log(Level.FINE, "Child balancer {0} deleted", key);
    }

    @Override
    public String toString() {
      return "Address = " + key
          + ", state = " + currentState
          + ", picker type: " + currentPicker.getClass()
          + ", lb: " + lb;
    }

    public final Object getKey() {
      return key;
    }

    @VisibleForTesting
    public final LoadBalancer getLb() {
      return lb;
    }

    @VisibleForTesting
    public final SubchannelPicker getCurrentPicker() {
      return currentPicker;
    }

    protected final Subchannel getSubchannels(PickSubchannelArgs args) {
      if (getCurrentPicker() == null) {
        return null;
      }
      return getCurrentPicker().pickSubchannel(args).getSubchannel();
    }

    public final ConnectivityState getCurrentState() {
      return currentState;
    }

    protected final void setCurrentState(ConnectivityState newState) {
      currentState = newState;
    }

    protected final void setCurrentPicker(SubchannelPicker newPicker) {
      currentPicker = newPicker;
    }

    public final EquivalentAddressGroup getEag() {
      if (resolvedAddresses == null || resolvedAddresses.getAddresses().isEmpty()) {
        return null;
      }
      return resolvedAddresses.getAddresses().get(0);
    }

    protected final void setResolvedAddresses(ResolvedAddresses newAddresses) {
      checkNotNull(newAddresses, "Missing address list for child");
      resolvedAddresses = newAddresses;
    }

    private Object getConfig() {
      return config;
    }

    @VisibleForTesting
    public final ResolvedAddresses getResolvedAddresses() {
      return resolvedAddresses;
    }

    /**
     * ChildLbStateHelper is the glue between ChildLbState and the helpers associated with the
     * petiole policy above and the PickFirstLoadBalancer's helper below.
     *
     * <p>The ChildLbState updates happen during updateBalancingState.  Otherwise, it is doing
     * simple forwarding.
     */
    protected class ChildLbStateHelper extends ForwardingLoadBalancerHelper {

      /**
       * Update current state and picker for this child and then use
       * {@link #updateOverallBalancingState()} for the parent LB.
       *
       * <p/>Override this if you don't want to automatically request a connection when in IDLE
       */
      @Override
      public void updateBalancingState(final ConnectivityState newState,
          final SubchannelPicker newPicker) {
        if (!childLbStates.containsKey(key)) {
          return;
        }

        currentState = newState;
        currentPicker = newPicker;
        // If we are already in the process of resolving addresses, the overall balancing state
        // will be updated at the end of it, and we don't need to trigger that update here.
        if (!resolvingAddresses) {
          if (newState == IDLE) {
            lb.requestConnection();
          }
          updateOverallBalancingState();
        }
      }

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

  /**
   * Endpoint is an optimization to quickly lookup and compare EquivalentAddressGroup address sets.
   * Ignores the attributes, orders the addresses in a deterministic manner and converts each
   * address into a string for easy comparison.  Also caches the hashcode.
   * Is used as a key for ChildLbState for most load balancers (ClusterManagerLB uses a String).
   */
  protected static class Endpoint {
    final String[] addrs;
    final int hashCode;

    public Endpoint(EquivalentAddressGroup eag) {
      checkNotNull(eag, "eag");

      addrs = new String[eag.getAddresses().size()];
      int i = 0;
      for (SocketAddress address : eag.getAddresses()) {
        addrs[i++] = address.toString();
      }
      Arrays.sort(addrs);

      hashCode = Arrays.hashCode(addrs);
    }

    @Override
    public int hashCode() {
      return hashCode;
    }

    @Override
    public boolean equals(Object other) {
      if (this == other) {
        return true;
      }
      if (other == null) {
        return false;
      }

      if (!(other instanceof Endpoint)) {
        return false;
      }
      Endpoint o = (Endpoint) other;
      if (o.hashCode != hashCode || o.addrs.length != addrs.length) {
        return false;
      }

      return Arrays.equals(o.addrs, this.addrs);
    }

    @Override
    public String toString() {
      return Arrays.toString(addrs);
    }
  }

  protected static class AcceptResolvedAddrRetVal {
    public final Status status;
    public final List<ChildLbState> removedChildren;

    public AcceptResolvedAddrRetVal(Status status, List<ChildLbState> removedChildren) {
      this.status = status;
      this.removedChildren = removedChildren;
    }
  }
}

1	/*
2	* Copyright 2023 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.util;
18
19	import static com.google.common.base.Preconditions.checkArgument;
20	import static com.google.common.base.Preconditions.checkNotNull;
21	import static io.grpc.ConnectivityState.CONNECTING;
22	import static io.grpc.ConnectivityState.IDLE;
23	import static io.grpc.ConnectivityState.READY;
24	import static io.grpc.ConnectivityState.SHUTDOWN;
25	import static io.grpc.ConnectivityState.TRANSIENT_FAILURE;
26
27	import com.google.common.annotations.VisibleForTesting;
28	import com.google.common.collect.ImmutableList;
29	import com.google.common.collect.ImmutableMap;
30	import io.grpc.Attributes;
31	import io.grpc.ConnectivityState;
32	import io.grpc.EquivalentAddressGroup;
33	import io.grpc.Internal;
34	import io.grpc.LoadBalancer;
35	import io.grpc.LoadBalancerProvider;
36	import io.grpc.Status;
37	import io.grpc.internal.PickFirstLoadBalancerProvider;
38	import java.net.SocketAddress;
39	import java.util.ArrayList;
40	import java.util.Arrays;
41	import java.util.Collection;
42	import java.util.Collections;
43	import java.util.HashMap;
44	import java.util.LinkedHashMap;
45	import java.util.List;
46	import java.util.Map;
47	import java.util.Set;
48	import java.util.logging.Level;
49	import java.util.logging.Logger;
50	import javax.annotation.Nullable;
51
52	/**
53	* A base load balancing policy for those policies which has multiple children such as
54	* ClusterManager or the petiole policies. For internal use only.
55	*/
56	@Internal
57	public abstract class MultiChildLoadBalancer extends LoadBalancer {
58
59	private static final Logger logger = Logger.getLogger(MultiChildLoadBalancer.class.getName());	1✔
60	private final Map<Object, ChildLbState> childLbStates = new LinkedHashMap<>();	1✔
61	private final Helper helper;
62	// Set to true if currently in the process of handling resolved addresses.
63	protected boolean resolvingAddresses;
64
65	protected final LoadBalancerProvider pickFirstLbProvider = new PickFirstLoadBalancerProvider();	1✔
66
67	protected ConnectivityState currentConnectivityState;
68
69
70	protected MultiChildLoadBalancer(Helper helper) {	1✔
71	this.helper = checkNotNull(helper, "helper");	1✔
72	logger.log(Level.FINE, "Created");	1✔
73	}	1✔
74
75	/**
76	* Using the state of all children will calculate the current connectivity state,
77	* update fields, generate a picker and then call
78	* {@link Helper#updateBalancingState(ConnectivityState, SubchannelPicker)}.
79	*/
80	protected abstract void updateOverallBalancingState();
81
82	/**
83	* Override to utilize parsing of the policy configuration or alternative helper/lb generation.
84	*/
85	protected Map<Object, ChildLbState> createChildLbMap(ResolvedAddresses resolvedAddresses) {
86	Map<Object, ChildLbState> childLbMap = new HashMap<>();	1✔
87	List<EquivalentAddressGroup> addresses = resolvedAddresses.getAddresses();	1✔
88	for (EquivalentAddressGroup eag : addresses) {	1✔
89	Endpoint endpoint = new Endpoint(eag); // keys need to be just addresses	1✔
90	ChildLbState existingChildLbState = childLbStates.get(endpoint);	1✔
91	if (existingChildLbState != null) {	1✔
92	childLbMap.put(endpoint, existingChildLbState);	1✔
93	} else {
94	childLbMap.put(endpoint,	1✔
95	createChildLbState(endpoint, null, getInitialPicker(), resolvedAddresses));	1✔
96	}
97	}	1✔
98	return childLbMap;	1✔
99	}
100
101	/**
102	* Override to create an instance of a subclass.
103	*/
104	protected ChildLbState createChildLbState(Object key, Object policyConfig,
105	SubchannelPicker initialPicker, ResolvedAddresses resolvedAddresses) {
106	return new ChildLbState(key, pickFirstLbProvider, policyConfig, initialPicker);	1✔
107	}
108
109	/**
110	* Override to completely replace the default logic or to do additional activities.
111	*/
112	@Override
113	public Status acceptResolvedAddresses(ResolvedAddresses resolvedAddresses) {
114	try {
115	resolvingAddresses = true;	1✔
116
117	// process resolvedAddresses to update children
118	AcceptResolvedAddrRetVal acceptRetVal = acceptResolvedAddressesInternal(resolvedAddresses);	1✔
119	if (!acceptRetVal.status.isOk()) {	1✔
120	return acceptRetVal.status;	1✔
121	}
122
123	// Update the picker and our connectivity state
124	updateOverallBalancingState();	1✔
125
126	// shutdown removed children
127	shutdownRemoved(acceptRetVal.removedChildren);	1✔
128	return acceptRetVal.status;	1✔
129	} finally {
130	resolvingAddresses = false;	1✔
131	}
132	}
133
134	/**
135	* Override this if your keys are not of type Endpoint.
136	* @param key Key to identify the ChildLbState
137	* @param resolvedAddresses list of addresses which include attributes
138	* @param childConfig a load balancing policy config. This field is optional.
139	* @return a fully loaded ResolvedAddresses object for the specified key
140	*/
141	protected ResolvedAddresses getChildAddresses(Object key, ResolvedAddresses resolvedAddresses,
142	Object childConfig) {
143	Endpoint endpointKey;
144	if (key instanceof EquivalentAddressGroup) {	1✔
145	endpointKey = new Endpoint((EquivalentAddressGroup) key);	×
146	} else {
147	checkArgument(key instanceof Endpoint, "key is wrong type");	1✔
148	endpointKey = (Endpoint) key;	1✔
149	}
150
151	// Retrieve the non-stripped version
152	EquivalentAddressGroup eagToUse = null;	1✔
153	for (EquivalentAddressGroup currEag : resolvedAddresses.getAddresses()) {	1✔
154	if (endpointKey.equals(new Endpoint(currEag))) {	1✔
155	eagToUse = currEag;	1✔
156	break;	1✔
157	}
158	}	1✔
159
160	checkNotNull(eagToUse, key + " no longer present in load balancer children");	1✔
161
162	return resolvedAddresses.toBuilder()	1✔
163	.setAddresses(Collections.singletonList(eagToUse))	1✔
164	.setAttributes(Attributes.newBuilder().set(IS_PETIOLE_POLICY, true).build())	1✔
165	.setLoadBalancingPolicyConfig(childConfig)	1✔
166	.build();	1✔
167	}
168
169	/**
170	* Handle the name resolution error.
171	*
172	* <p/>Override if you need special handling.
173	*/
174	@Override
175	public void handleNameResolutionError(Status error) {
176	if (currentConnectivityState != READY) {	1✔
177	helper.updateBalancingState(TRANSIENT_FAILURE, getErrorPicker(error));	1✔
178	}
179	}	1✔
180
181	/**
182	* Handle the name resolution error only for the specified child.
183	*
184	* <p/>Override if you need special handling.
185	*/
186	protected void handleNameResolutionError(ChildLbState child, Status error) {
187	child.lb.handleNameResolutionError(error);	1✔
188	}	1✔
189
190	/**
191	* Creates a picker representing the state before any connections have been established.
192	*
193	* <p/>Override to produce a custom picker.
194	*/
195	protected SubchannelPicker getInitialPicker() {
196	return new FixedResultPicker(PickResult.withNoResult());	1✔
197	}
198
199	/**
200	* Creates a new picker representing an error status.
201	*
202	* <p/>Override to produce a custom picker when there are errors.
203	*/
204	protected SubchannelPicker getErrorPicker(Status error) {
205	return new FixedResultPicker(PickResult.withError(error));	1✔
206	}
207
208	@Override
209	public void shutdown() {
210	logger.log(Level.FINE, "Shutdown");	1✔
211	for (ChildLbState state : childLbStates.values()) {	1✔
212	state.shutdown();	1✔
213	}	1✔
214	childLbStates.clear();	1✔
215	}	1✔
216
217	/**
218	* This does the work to update the child map and calculate which children have been removed.
219	* You must call {@link #updateOverallBalancingState} to update the picker
220	* and call {@link #shutdownRemoved(List)} to shutdown the endpoints that have been removed.
221	*/
222	protected final AcceptResolvedAddrRetVal acceptResolvedAddressesInternal(
223	ResolvedAddresses resolvedAddresses) {
224	logger.log(Level.FINE, "Received resolution result: {0}", resolvedAddresses);	1✔
225
226	// Subclass handles any special manipulation to create appropriate types of keyed ChildLbStates
227	Map<Object, ChildLbState> newChildren = createChildLbMap(resolvedAddresses);	1✔
228
229	// Handle error case
230	if (newChildren.isEmpty()) {	1✔
231	Status unavailableStatus = Status.UNAVAILABLE.withDescription(	1✔
232	"NameResolver returned no usable address. " + resolvedAddresses);
233	handleNameResolutionError(unavailableStatus);	1✔
234	return new AcceptResolvedAddrRetVal(unavailableStatus, null);	1✔
235	}
236
237	addMissingChildren(newChildren);	1✔
238
239	updateChildrenWithResolvedAddresses(resolvedAddresses, newChildren);	1✔
240
241	return new AcceptResolvedAddrRetVal(Status.OK, getRemovedChildren(newChildren.keySet()));	1✔
242	}
243
244	protected final void addMissingChildren(Map<Object, ChildLbState> newChildren) {
245	// Do adds and identify reused children
246	for (Map.Entry<Object, ChildLbState> entry : newChildren.entrySet()) {	1✔
247	final Object key = entry.getKey();	1✔
248	if (!childLbStates.containsKey(key)) {	1✔
249	childLbStates.put(key, entry.getValue());	1✔
250	}
251	}	1✔
252	}	1✔
253
254	protected final void updateChildrenWithResolvedAddresses(ResolvedAddresses resolvedAddresses,
255	Map<Object, ChildLbState> newChildren) {
256	for (Map.Entry<Object, ChildLbState> entry : newChildren.entrySet()) {	1✔
257	Object childConfig = entry.getValue().getConfig();	1✔
258	ChildLbState childLbState = childLbStates.get(entry.getKey());	1✔
259	ResolvedAddresses childAddresses =	1✔
260	getChildAddresses(entry.getKey(), resolvedAddresses, childConfig);	1✔
261	childLbState.setResolvedAddresses(childAddresses); // update child	1✔
262	childLbState.lb.handleResolvedAddresses(childAddresses); // update child LB	1✔
263	}	1✔
264	}	1✔
265
266	/**
267	* Identifies which children have been removed (are not part of the newChildKeys).
268	*/
269	protected final List<ChildLbState> getRemovedChildren(Set<Object> newChildKeys) {
270	List<ChildLbState> removedChildren = new ArrayList<>();	1✔
271	// Do removals
272	for (Object key : ImmutableList.copyOf(childLbStates.keySet())) {	1✔
273	if (!newChildKeys.contains(key)) {	1✔
274	ChildLbState childLbState = childLbStates.remove(key);	1✔
275	removedChildren.add(childLbState);	1✔
276	}
277	}	1✔
278	return removedChildren;	1✔
279	}
280
281	protected final void shutdownRemoved(List<ChildLbState> removedChildren) {
282	// Do shutdowns after updating picker to reduce the chance of failing an RPC by picking a
283	// subchannel that has been shutdown.
284	for (ChildLbState childLbState : removedChildren) {	1✔
285	childLbState.shutdown();	1✔
286	}	1✔
287	}	1✔
288
289	@Nullable
290	protected static ConnectivityState aggregateState(
291	@Nullable ConnectivityState overallState, ConnectivityState childState) {
292	if (overallState == null) {	1✔
293	return childState;	1✔
294	}
295	if (overallState == READY \|\| childState == READY) {	1✔
296	return READY;	1✔
297	}
298	if (overallState == CONNECTING \|\| childState == CONNECTING) {	1✔
299	return CONNECTING;	1✔
300	}
301	if (overallState == IDLE \|\| childState == IDLE) {	1✔
302	return IDLE;	×
303	}
304	return overallState;	1✔
305	}
306
307	protected final Helper getHelper() {
308	return helper;	1✔
309	}
310
311	@VisibleForTesting
312	public final ImmutableMap<Object, ChildLbState> getImmutableChildMap() {
313	return ImmutableMap.copyOf(childLbStates);	1✔
314	}
315
316	@VisibleForTesting
317	public final Collection<ChildLbState> getChildLbStates() {
318	return childLbStates.values();	1✔
319	}
320
321	@VisibleForTesting
322	public final ChildLbState getChildLbState(Object key) {
323	if (key == null) {	1✔
324	return null;	×
325	}
326	if (key instanceof EquivalentAddressGroup) {	1✔
327	key = new Endpoint((EquivalentAddressGroup) key);	1✔
328	}
329	return childLbStates.get(key);	1✔
330	}
331
332	@VisibleForTesting
333	public final ChildLbState getChildLbStateEag(EquivalentAddressGroup eag) {
334	return getChildLbState(new Endpoint(eag));	1✔
335	}
336
337	/**
338	* Filters out non-ready child load balancers (subchannels).
339	*/
340	protected final List<ChildLbState> getReadyChildren() {
341	List<ChildLbState> activeChildren = new ArrayList<>();	1✔
342	for (ChildLbState child : getChildLbStates()) {	1✔
343	if (child.getCurrentState() == READY) {	1✔
344	activeChildren.add(child);	1✔
345	}
346	}	1✔
347	return activeChildren;	1✔
348	}
349
350	/**
351	* This represents the state of load balancer children. Each endpoint (represented by an
352	* EquivalentAddressGroup or EDS string) will have a separate ChildLbState which in turn will
353	* have a single child LoadBalancer created from the provided factory.
354	*
355	* <p>A ChildLbStateHelper is the glue between ChildLbState and the helpers associated with the
356	* petiole policy above and the PickFirstLoadBalancer's helper below.
357	*
358	* <p>If you wish to store additional state information related to each subchannel, then extend
359	* this class.
360	*/
361	public class ChildLbState {
362	private final Object key;
363	private ResolvedAddresses resolvedAddresses;
364	private final Object config;
365
366	private final LoadBalancer lb;
367	private ConnectivityState currentState;
368	private SubchannelPicker currentPicker;
369
370	public ChildLbState(Object key, LoadBalancer.Factory policyFactory, Object childConfig,
371	SubchannelPicker initialPicker) {	1✔
372	this.key = key;	1✔
373	this.currentPicker = initialPicker;	1✔
374	this.config = childConfig;	1✔
375	this.lb = policyFactory.newLoadBalancer(createChildHelper());	1✔
376	this.currentState = CONNECTING;	1✔
377	}	1✔
378
379	protected ChildLbStateHelper createChildHelper() {
380	return new ChildLbStateHelper();	1✔
381	}
382
383	/**
384	* Override for unique behavior such as delayed shutdowns of subchannels.
385	*/
386	protected void shutdown() {
387	lb.shutdown();	1✔
388	this.currentState = SHUTDOWN;	1✔
389	logger.log(Level.FINE, "Child balancer {0} deleted", key);	1✔
390	}	1✔
391
392	@Override
393	public String toString() {
394	return "Address = " + key	1✔
395	+ ", state = " + currentState
396	+ ", picker type: " + currentPicker.getClass()	1✔
397	+ ", lb: " + lb;
398	}
399
400	public final Object getKey() {
401	return key;	1✔
402	}
403
404	@VisibleForTesting
405	public final LoadBalancer getLb() {
406	return lb;	1✔
407	}
408
409	@VisibleForTesting
410	public final SubchannelPicker getCurrentPicker() {
411	return currentPicker;	1✔
412	}
413
414	protected final Subchannel getSubchannels(PickSubchannelArgs args) {
415	if (getCurrentPicker() == null) {	×
416	return null;	×
417	}
418	return getCurrentPicker().pickSubchannel(args).getSubchannel();	×
419	}
420
421	public final ConnectivityState getCurrentState() {
422	return currentState;	1✔
423	}
424
425	protected final void setCurrentState(ConnectivityState newState) {
426	currentState = newState;	1✔
427	}	1✔
428
429	protected final void setCurrentPicker(SubchannelPicker newPicker) {
430	currentPicker = newPicker;	1✔
431	}	1✔
432
433	public final EquivalentAddressGroup getEag() {
434	if (resolvedAddresses == null \|\| resolvedAddresses.getAddresses().isEmpty()) {	1✔
435	return null;	×
436	}
437	return resolvedAddresses.getAddresses().get(0);	1✔
438	}
439
440	protected final void setResolvedAddresses(ResolvedAddresses newAddresses) {
441	checkNotNull(newAddresses, "Missing address list for child");	1✔
442	resolvedAddresses = newAddresses;	1✔
443	}	1✔
444
445	private Object getConfig() {
446	return config;	1✔
447	}
448
449	@VisibleForTesting
450	public final ResolvedAddresses getResolvedAddresses() {
451	return resolvedAddresses;	1✔
452	}
453
454	/**
455	* ChildLbStateHelper is the glue between ChildLbState and the helpers associated with the
456	* petiole policy above and the PickFirstLoadBalancer's helper below.
457	*
458	* <p>The ChildLbState updates happen during updateBalancingState. Otherwise, it is doing
459	* simple forwarding.
460	*/
461	protected class ChildLbStateHelper extends ForwardingLoadBalancerHelper {	1✔
462
463	/**
464	* Update current state and picker for this child and then use
465	* {@link #updateOverallBalancingState()} for the parent LB.
466	*
467	* <p/>Override this if you don't want to automatically request a connection when in IDLE
468	*/
469	@Override
470	public void updateBalancingState(final ConnectivityState newState,
471	final SubchannelPicker newPicker) {
472	if (!childLbStates.containsKey(key)) {	1✔
473	return;	×
474	}
475
476	currentState = newState;	1✔
477	currentPicker = newPicker;	1✔
478	// If we are already in the process of resolving addresses, the overall balancing state
479	// will be updated at the end of it, and we don't need to trigger that update here.
480	if (!resolvingAddresses) {	1✔
481	if (newState == IDLE) {	1✔
482	lb.requestConnection();	1✔
483	}
484	updateOverallBalancingState();	1✔
485	}
486	}	1✔
487
488	@Override
489	protected Helper delegate() {
490	return helper;	1✔
491	}
492	}
493	}
494
495	/**
496	* Endpoint is an optimization to quickly lookup and compare EquivalentAddressGroup address sets.
497	* Ignores the attributes, orders the addresses in a deterministic manner and converts each
498	* address into a string for easy comparison. Also caches the hashcode.
499	* Is used as a key for ChildLbState for most load balancers (ClusterManagerLB uses a String).
500	*/
501	protected static class Endpoint {
502	final String[] addrs;
503	final int hashCode;
504
505	public Endpoint(EquivalentAddressGroup eag) {	1✔
506	checkNotNull(eag, "eag");	1✔
507
508	addrs = new String[eag.getAddresses().size()];	1✔
509	int i = 0;	1✔
510	for (SocketAddress address : eag.getAddresses()) {	1✔
511	addrs[i++] = address.toString();	1✔
512	}	1✔
513	Arrays.sort(addrs);	1✔
514
515	hashCode = Arrays.hashCode(addrs);	1✔
516	}	1✔
517
518	@Override
519	public int hashCode() {
520	return hashCode;	1✔
521	}
522
523	@Override
524	public boolean equals(Object other) {
525	if (this == other) {	1✔
526	return true;	×
527	}
528	if (other == null) {	1✔
529	return false;	×
530	}
531
532	if (!(other instanceof Endpoint)) {	1✔
533	return false;	×
534	}
535	Endpoint o = (Endpoint) other;	1✔
536	if (o.hashCode != hashCode \|\| o.addrs.length != addrs.length) {	1✔
537	return false;	1✔
538	}
539
540	return Arrays.equals(o.addrs, this.addrs);	1✔
541	}
542
543	@Override
544	public String toString() {
545	return Arrays.toString(addrs);	1✔
546	}
547	}
548
549	protected static class AcceptResolvedAddrRetVal {
550	public final Status status;
551	public final List<ChildLbState> removedChildren;
552
553	public AcceptResolvedAddrRetVal(Status status, List<ChildLbState> removedChildren) {	1✔
554	this.status = status;	1✔
555	this.removedChildren = removedChildren;	1✔
556	}	1✔
557	}
558	}

grpc / grpc-java / #19167

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