#19440

Committed 29 Aug 2024 03:04PM UTC coverage: 84.491% (-0.002%) from 84.493%

Build # #19440

Build Type

push

github

Committed by

ejona86

Commit Message

util: Remove child policy config from MultiChildLB state

The child policy config should be refreshed every address update, so it
shouldn't be stored in the ChildLbState. In addition, none of the
current usages actually used what was stored in the ChildLbState in a
meaningful way (it was always null).

ResolvedAddresses was also removed from createChildLbState(), as nothing
in it should be needed for creation; it varies over time and the values
passed at creation are immutable.

Run Details

33402 of 39533 relevant lines covered (84.49%)

0.84 hits per line

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

99.17

/../xds/src/main/java/io/grpc/xds/ClusterManagerLoadBalancer.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 static io.grpc.ConnectivityState.TRANSIENT_FAILURE;

import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.MoreObjects;
import io.grpc.ConnectivityState;
import io.grpc.InternalLogId;
import io.grpc.LoadBalancer;
import io.grpc.Status;
import io.grpc.SynchronizationContext;
import io.grpc.SynchronizationContext.ScheduledHandle;
import io.grpc.util.GracefulSwitchLoadBalancer;
import io.grpc.util.MultiChildLoadBalancer;
import io.grpc.xds.ClusterManagerLoadBalancerProvider.ClusterManagerConfig;
import io.grpc.xds.client.XdsLogger;
import io.grpc.xds.client.XdsLogger.XdsLogLevel;
import java.util.HashMap;
import java.util.Map;
import java.util.Map.Entry;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import javax.annotation.Nullable;

/**
 * The top-level load balancing policy for use in XDS.
 * This policy does not immediately delete its children.  Instead, it marks them deactivated
 * and starts a timer for deletion.  If a subsequent address update restores the child, then it is
 * simply reactivated instead of built from scratch.  This is necessary because XDS can frequently
 * remove and then add back a server as machines are rebooted or repurposed for load management.
 *
 * <p>Note that this LB does not automatically reconnect children who go into IDLE status
 */
class ClusterManagerLoadBalancer extends MultiChildLoadBalancer {

  // 15 minutes is long enough for a reboot and the services to restart while not so long that
  // many children are waiting for cleanup.
  @VisibleForTesting
  public static final int DELAYED_CHILD_DELETION_TIME_MINUTES = 15;
  protected final SynchronizationContext syncContext;
  private final ScheduledExecutorService timeService;
  private final XdsLogger logger;
  private ResolvedAddresses lastResolvedAddresses;

  ClusterManagerLoadBalancer(Helper helper) {
    super(helper);
    this.syncContext = checkNotNull(helper.getSynchronizationContext(), "syncContext");
    this.timeService = checkNotNull(helper.getScheduledExecutorService(), "timeService");
    logger = XdsLogger.withLogId(
        InternalLogId.allocate("cluster_manager-lb", helper.getAuthority()));

    logger.log(XdsLogLevel.INFO, "Created");
  }

  @Override
  protected ResolvedAddresses getChildAddresses(Object key, ResolvedAddresses resolvedAddresses) {
    ClusterManagerConfig config = (ClusterManagerConfig)
        resolvedAddresses.getLoadBalancingPolicyConfig();
    Object childConfig = config.childPolicies.get(key);
    return resolvedAddresses.toBuilder().setLoadBalancingPolicyConfig(childConfig).build();
  }

  @Override
  protected Map<Object, ChildLbState> createChildLbMap(ResolvedAddresses resolvedAddresses) {
    ClusterManagerConfig config = (ClusterManagerConfig)
        resolvedAddresses.getLoadBalancingPolicyConfig();
    Map<Object, ChildLbState> newChildPolicies = new HashMap<>();
    if (config != null) {
      for (String key : config.childPolicies.keySet()) {
        ChildLbState child = getChildLbState(key);
        if (child == null) {
          child = new ClusterManagerLbState(key, GracefulSwitchLoadBalancerFactory.INSTANCE);
        }
        newChildPolicies.put(key, child);
      }
    }
    logger.log(
        XdsLogLevel.INFO,
        "Received cluster_manager lb config: child names={0}", newChildPolicies.keySet());
    return newChildPolicies;
  }

  /**
   * This is like the parent except that it doesn't shutdown the removed children since we want that
   * to be done by the timer.
   */
  @Override
  public Status acceptResolvedAddresses(ResolvedAddresses resolvedAddresses) {
    if (lastResolvedAddresses != null) {
      // Handle deactivated children
      ClusterManagerConfig config = (ClusterManagerConfig)
          resolvedAddresses.getLoadBalancingPolicyConfig();
      ClusterManagerConfig lastConfig = (ClusterManagerConfig)
          lastResolvedAddresses.getLoadBalancingPolicyConfig();
      Map<String, Object> adjChildPolicies = new HashMap<>(config.childPolicies);
      for (Entry<String, Object> entry : lastConfig.childPolicies.entrySet()) {
        ClusterManagerLbState state = (ClusterManagerLbState) getChildLbState(entry.getKey());
        if (adjChildPolicies.containsKey(entry.getKey())) {
          if (state.deletionTimer != null) {
            state.reactivateChild();
          }
        } else if (state != null) {
          adjChildPolicies.put(entry.getKey(), entry.getValue());
          if (state.deletionTimer == null) {
            state.deactivateChild();
          }
        }
      }
      config = new ClusterManagerConfig(adjChildPolicies);
      resolvedAddresses =
          resolvedAddresses.toBuilder().setLoadBalancingPolicyConfig(config).build();
    }
    lastResolvedAddresses = resolvedAddresses;
    return super.acceptResolvedAddresses(resolvedAddresses);
  }

  /**
   * Using the state of all children will calculate the current connectivity state,
   * update currentConnectivityState, generate a picker and then call
   * {@link Helper#updateBalancingState(ConnectivityState, SubchannelPicker)}.
   */
  @Override
  protected void updateOverallBalancingState() {
    ConnectivityState overallState = null;
    final Map<Object, SubchannelPicker> childPickers = new HashMap<>();
    for (ChildLbState childLbState : getChildLbStates()) {
      if (((ClusterManagerLbState) childLbState).deletionTimer != null) {
        continue;
      }
      childPickers.put(childLbState.getKey(), childLbState.getCurrentPicker());
      overallState = aggregateState(overallState, childLbState.getCurrentState());
    }

    if (overallState != null) {
      getHelper().updateBalancingState(overallState, getSubchannelPicker(childPickers));
      currentConnectivityState = overallState;
    }
  }

  protected SubchannelPicker getSubchannelPicker(Map<Object, SubchannelPicker> childPickers) {
    return new SubchannelPicker() {
      @Override
      public PickResult pickSubchannel(PickSubchannelArgs args) {
        String clusterName =
            args.getCallOptions().getOption(XdsNameResolver.CLUSTER_SELECTION_KEY);
        SubchannelPicker childPicker = childPickers.get(clusterName);
        if (childPicker == null) {
          return
              PickResult.withError(
                  Status.UNAVAILABLE.withDescription("CDS encountered error: unable to find "
                      + "available subchannel for cluster " + clusterName));
        }
        return childPicker.pickSubchannel(args);
      }

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

  @Override
  public void handleNameResolutionError(Status error) {
    logger.log(XdsLogLevel.WARNING, "Received name resolution error: {0}", error);
    boolean gotoTransientFailure = true;
    for (ChildLbState state : getChildLbStates()) {
      if (((ClusterManagerLbState) state).deletionTimer == null) {
        gotoTransientFailure = false;
        state.getLb().handleNameResolutionError(error);
      }
    }
    if (gotoTransientFailure) {
      getHelper().updateBalancingState(
          TRANSIENT_FAILURE, new FixedResultPicker(PickResult.withError(error)));
    }
  }

  /**
   * This differs from the base class in the use of the deletion timer.  When it is deactivated,
   * rather than immediately calling shutdown it starts a timer.  If shutdown or reactivate
   * are called before the timer fires, the timer is canceled.  Otherwise, time timer calls shutdown
   * and removes the child from the petiole policy when it is triggered.
   */
  private class ClusterManagerLbState extends ChildLbState {
    @Nullable
    ScheduledHandle deletionTimer;

    public ClusterManagerLbState(Object key, LoadBalancer.Factory policyFactory) {
      super(key, policyFactory);
    }

    @Override
    protected ChildLbStateHelper createChildHelper() {
      return new ClusterManagerChildHelper();
    }

    @Override
    protected void shutdown() {
      if (deletionTimer != null) {
        deletionTimer.cancel();
        deletionTimer = null;
      }
      super.shutdown();
    }

    void reactivateChild() {
      assert deletionTimer != null;
      deletionTimer.cancel();
      deletionTimer = null;
      logger.log(XdsLogLevel.DEBUG, "Child balancer {0} reactivated", getKey());
    }

    void deactivateChild() {
      assert deletionTimer == null;

      class DeletionTask implements Runnable {

        @Override
        public void run() {
          ClusterManagerConfig config = (ClusterManagerConfig)
              lastResolvedAddresses.getLoadBalancingPolicyConfig();
          Map<String, Object> childPolicies = new HashMap<>(config.childPolicies);
          Object removed = childPolicies.remove(getKey());
          assert removed != null;
          config = new ClusterManagerConfig(childPolicies);
          lastResolvedAddresses =
              lastResolvedAddresses.toBuilder().setLoadBalancingPolicyConfig(config).build();
          acceptResolvedAddresses(lastResolvedAddresses);
        }
      }

      deletionTimer =
          syncContext.schedule(
              new DeletionTask(),
              DELAYED_CHILD_DELETION_TIME_MINUTES,
              TimeUnit.MINUTES,
              timeService);
      logger.log(XdsLogLevel.DEBUG, "Child balancer {0} deactivated", getKey());
    }

    private class ClusterManagerChildHelper extends ChildLbStateHelper {
      @Override
      public void updateBalancingState(final ConnectivityState newState,
                                       final SubchannelPicker 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 (getChildLbState(getKey()) == null) {
          return;
        }

        // Subchannel picker and state are saved, but will only be propagated to the channel
        // when the child instance exits deactivated state.
        setCurrentState(newState);
        setCurrentPicker(newPicker);
        if (deletionTimer == null && !resolvingAddresses) {
          updateOverallBalancingState();
        }
      }
    }
  }

  static final class GracefulSwitchLoadBalancerFactory extends LoadBalancer.Factory {
    static final LoadBalancer.Factory INSTANCE = new GracefulSwitchLoadBalancerFactory();

    @Override
    public LoadBalancer newLoadBalancer(LoadBalancer.Helper helper) {
      return new GracefulSwitchLoadBalancer(helper);
    }
  }
}

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	import static io.grpc.ConnectivityState.TRANSIENT_FAILURE;
21
22	import com.google.common.annotations.VisibleForTesting;
23	import com.google.common.base.MoreObjects;
24	import io.grpc.ConnectivityState;
25	import io.grpc.InternalLogId;
26	import io.grpc.LoadBalancer;
27	import io.grpc.Status;
28	import io.grpc.SynchronizationContext;
29	import io.grpc.SynchronizationContext.ScheduledHandle;
30	import io.grpc.util.GracefulSwitchLoadBalancer;
31	import io.grpc.util.MultiChildLoadBalancer;
32	import io.grpc.xds.ClusterManagerLoadBalancerProvider.ClusterManagerConfig;
33	import io.grpc.xds.client.XdsLogger;
34	import io.grpc.xds.client.XdsLogger.XdsLogLevel;
35	import java.util.HashMap;
36	import java.util.Map;
37	import java.util.Map.Entry;
38	import java.util.concurrent.ScheduledExecutorService;
39	import java.util.concurrent.TimeUnit;
40	import javax.annotation.Nullable;
41
42	/**
43	* The top-level load balancing policy for use in XDS.
44	* This policy does not immediately delete its children. Instead, it marks them deactivated
45	* and starts a timer for deletion. If a subsequent address update restores the child, then it is
46	* simply reactivated instead of built from scratch. This is necessary because XDS can frequently
47	* remove and then add back a server as machines are rebooted or repurposed for load management.
48	*
49	* <p>Note that this LB does not automatically reconnect children who go into IDLE status
50	*/
51	class ClusterManagerLoadBalancer extends MultiChildLoadBalancer {
52
53	// 15 minutes is long enough for a reboot and the services to restart while not so long that
54	// many children are waiting for cleanup.
55	@VisibleForTesting
56	public static final int DELAYED_CHILD_DELETION_TIME_MINUTES = 15;
57	protected final SynchronizationContext syncContext;
58	private final ScheduledExecutorService timeService;
59	private final XdsLogger logger;
60	private ResolvedAddresses lastResolvedAddresses;
61
62	ClusterManagerLoadBalancer(Helper helper) {
63	super(helper);	1✔
64	this.syncContext = checkNotNull(helper.getSynchronizationContext(), "syncContext");	1✔
65	this.timeService = checkNotNull(helper.getScheduledExecutorService(), "timeService");	1✔
66	logger = XdsLogger.withLogId(	1✔
67	InternalLogId.allocate("cluster_manager-lb", helper.getAuthority()));	1✔
68
69	logger.log(XdsLogLevel.INFO, "Created");	1✔
70	}	1✔
71
72	@Override
73	protected ResolvedAddresses getChildAddresses(Object key, ResolvedAddresses resolvedAddresses) {
74	ClusterManagerConfig config = (ClusterManagerConfig)	1✔
75	resolvedAddresses.getLoadBalancingPolicyConfig();	1✔
76	Object childConfig = config.childPolicies.get(key);	1✔
77	return resolvedAddresses.toBuilder().setLoadBalancingPolicyConfig(childConfig).build();	1✔
78	}
79
80	@Override
81	protected Map<Object, ChildLbState> createChildLbMap(ResolvedAddresses resolvedAddresses) {
82	ClusterManagerConfig config = (ClusterManagerConfig)	1✔
83	resolvedAddresses.getLoadBalancingPolicyConfig();	1✔
84	Map<Object, ChildLbState> newChildPolicies = new HashMap<>();	1✔
85	if (config != null) {	1✔
86	for (String key : config.childPolicies.keySet()) {	1✔
87	ChildLbState child = getChildLbState(key);	1✔
88	if (child == null) {	1✔
89	child = new ClusterManagerLbState(key, GracefulSwitchLoadBalancerFactory.INSTANCE);	1✔
90	}
91	newChildPolicies.put(key, child);	1✔
92	}	1✔
93	}
94	logger.log(	1✔
95	XdsLogLevel.INFO,
96	"Received cluster_manager lb config: child names={0}", newChildPolicies.keySet());	1✔
97	return newChildPolicies;	1✔
98	}
99
100	/**
101	* This is like the parent except that it doesn't shutdown the removed children since we want that
102	* to be done by the timer.
103	*/
104	@Override
105	public Status acceptResolvedAddresses(ResolvedAddresses resolvedAddresses) {
106	if (lastResolvedAddresses != null) {	1✔
107	// Handle deactivated children
108	ClusterManagerConfig config = (ClusterManagerConfig)	1✔
109	resolvedAddresses.getLoadBalancingPolicyConfig();	1✔
110	ClusterManagerConfig lastConfig = (ClusterManagerConfig)	1✔
111	lastResolvedAddresses.getLoadBalancingPolicyConfig();	1✔
112	Map<String, Object> adjChildPolicies = new HashMap<>(config.childPolicies);	1✔
113	for (Entry<String, Object> entry : lastConfig.childPolicies.entrySet()) {	1✔
114	ClusterManagerLbState state = (ClusterManagerLbState) getChildLbState(entry.getKey());	1✔
115	if (adjChildPolicies.containsKey(entry.getKey())) {	1✔
116	if (state.deletionTimer != null) {	1✔
117	state.reactivateChild();	1✔
118	}
119	} else if (state != null) {	1✔
120	adjChildPolicies.put(entry.getKey(), entry.getValue());	1✔
121	if (state.deletionTimer == null) {	1✔
122	state.deactivateChild();	1✔
123	}
124	}
125	}	1✔
126	config = new ClusterManagerConfig(adjChildPolicies);	1✔
127	resolvedAddresses =	1✔
128	resolvedAddresses.toBuilder().setLoadBalancingPolicyConfig(config).build();	1✔
129	}
130	lastResolvedAddresses = resolvedAddresses;	1✔
131	return super.acceptResolvedAddresses(resolvedAddresses);	1✔
132	}
133
134	/**
135	* Using the state of all children will calculate the current connectivity state,
136	* update currentConnectivityState, generate a picker and then call
137	* {@link Helper#updateBalancingState(ConnectivityState, SubchannelPicker)}.
138	*/
139	@Override
140	protected void updateOverallBalancingState() {
141	ConnectivityState overallState = null;	1✔
142	final Map<Object, SubchannelPicker> childPickers = new HashMap<>();	1✔
143	for (ChildLbState childLbState : getChildLbStates()) {	1✔
144	if (((ClusterManagerLbState) childLbState).deletionTimer != null) {	1✔
145	continue;	1✔
146	}
147	childPickers.put(childLbState.getKey(), childLbState.getCurrentPicker());	1✔
148	overallState = aggregateState(overallState, childLbState.getCurrentState());	1✔
149	}	1✔
150
151	if (overallState != null) {	1✔
152	getHelper().updateBalancingState(overallState, getSubchannelPicker(childPickers));	1✔
153	currentConnectivityState = overallState;	1✔
154	}
155	}	1✔
156
157	protected SubchannelPicker getSubchannelPicker(Map<Object, SubchannelPicker> childPickers) {
158	return new SubchannelPicker() {	1✔
159	@Override
160	public PickResult pickSubchannel(PickSubchannelArgs args) {
161	String clusterName =	1✔
162	args.getCallOptions().getOption(XdsNameResolver.CLUSTER_SELECTION_KEY);	1✔
163	SubchannelPicker childPicker = childPickers.get(clusterName);	1✔
164	if (childPicker == null) {	1✔
165	return	1✔
166	PickResult.withError(	1✔
167	Status.UNAVAILABLE.withDescription("CDS encountered error: unable to find "	1✔
168	+ "available subchannel for cluster " + clusterName));
169	}
170	return childPicker.pickSubchannel(args);	1✔
171	}
172
173	@Override
174	public String toString() {
175	return MoreObjects.toStringHelper(this).add("pickers", childPickers).toString();	×
176	}
177	};
178	}
179
180	@Override
181	public void handleNameResolutionError(Status error) {
182	logger.log(XdsLogLevel.WARNING, "Received name resolution error: {0}", error);	1✔
183	boolean gotoTransientFailure = true;	1✔
184	for (ChildLbState state : getChildLbStates()) {	1✔
185	if (((ClusterManagerLbState) state).deletionTimer == null) {	1✔
186	gotoTransientFailure = false;	1✔
187	state.getLb().handleNameResolutionError(error);	1✔
188	}
189	}	1✔
190	if (gotoTransientFailure) {	1✔
191	getHelper().updateBalancingState(	1✔
192	TRANSIENT_FAILURE, new FixedResultPicker(PickResult.withError(error)));	1✔
193	}
194	}	1✔
195
196	/**
197	* This differs from the base class in the use of the deletion timer. When it is deactivated,
198	* rather than immediately calling shutdown it starts a timer. If shutdown or reactivate
199	* are called before the timer fires, the timer is canceled. Otherwise, time timer calls shutdown
200	* and removes the child from the petiole policy when it is triggered.
201	*/
202	private class ClusterManagerLbState extends ChildLbState {	1✔
203	@Nullable
204	ScheduledHandle deletionTimer;
205
206	public ClusterManagerLbState(Object key, LoadBalancer.Factory policyFactory) {	1✔
207	super(key, policyFactory);	1✔
208	}	1✔
209
210	@Override
211	protected ChildLbStateHelper createChildHelper() {
212	return new ClusterManagerChildHelper();	1✔
213	}
214
215	@Override
216	protected void shutdown() {
217	if (deletionTimer != null) {	1✔
218	deletionTimer.cancel();	1✔
219	deletionTimer = null;	1✔
220	}
221	super.shutdown();	1✔
222	}	1✔
223
224	void reactivateChild() {
225	assert deletionTimer != null;	1✔
226	deletionTimer.cancel();	1✔
227	deletionTimer = null;	1✔
228	logger.log(XdsLogLevel.DEBUG, "Child balancer {0} reactivated", getKey());	1✔
229	}	1✔
230
231	void deactivateChild() {
232	assert deletionTimer == null;	1✔
233
234	class DeletionTask implements Runnable {	1✔
235
236	@Override
237	public void run() {
238	ClusterManagerConfig config = (ClusterManagerConfig)	1✔
239	lastResolvedAddresses.getLoadBalancingPolicyConfig();	1✔
240	Map<String, Object> childPolicies = new HashMap<>(config.childPolicies);	1✔
241	Object removed = childPolicies.remove(getKey());	1✔
242	assert removed != null;	1✔
243	config = new ClusterManagerConfig(childPolicies);	1✔
244	lastResolvedAddresses =	1✔
245	lastResolvedAddresses.toBuilder().setLoadBalancingPolicyConfig(config).build();	1✔
246	acceptResolvedAddresses(lastResolvedAddresses);	1✔
247	}	1✔
248	}
249
250	deletionTimer =	1✔
251	syncContext.schedule(	1✔
252	new DeletionTask(),
253	DELAYED_CHILD_DELETION_TIME_MINUTES,
254	TimeUnit.MINUTES,
255	timeService);	1✔
256	logger.log(XdsLogLevel.DEBUG, "Child balancer {0} deactivated", getKey());	1✔
257	}	1✔
258
259	private class ClusterManagerChildHelper extends ChildLbStateHelper {	1✔
260	@Override
261	public void updateBalancingState(final ConnectivityState newState,
262	final SubchannelPicker newPicker) {
263	// If we are already in the process of resolving addresses, the overall balancing state
264	// will be updated at the end of it, and we don't need to trigger that update here.
265	if (getChildLbState(getKey()) == null) {	1✔
266	return;	1✔
267	}
268
269	// Subchannel picker and state are saved, but will only be propagated to the channel
270	// when the child instance exits deactivated state.
271	setCurrentState(newState);	1✔
272	setCurrentPicker(newPicker);	1✔
273	if (deletionTimer == null && !resolvingAddresses) {	1✔
274	updateOverallBalancingState();	1✔
275	}
276	}	1✔
277	}
278	}
279
280	static final class GracefulSwitchLoadBalancerFactory extends LoadBalancer.Factory {	1✔
281	static final LoadBalancer.Factory INSTANCE = new GracefulSwitchLoadBalancerFactory();	1✔
282
283	@Override
284	public LoadBalancer newLoadBalancer(LoadBalancer.Helper helper) {
285	return new GracefulSwitchLoadBalancer(helper);	1✔
286	}
287	}
288	}

grpc / grpc-java / #19440

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