#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

99.13

/../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.LoadBalancerProvider;
import io.grpc.Status;
import io.grpc.SynchronizationContext;
import io.grpc.SynchronizationContext.ScheduledHandle;
import io.grpc.internal.ServiceConfigUtil.PolicySelection;
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,
      Object childConfig) {
    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 (Entry<String, PolicySelection> entry : config.childPolicies.entrySet()) {
        ChildLbState child = getChildLbState(entry.getKey());
        if (child == null) {
          child = new ClusterManagerLbState(entry.getKey(),
              entry.getValue().getProvider(), entry.getValue().getConfig(), getInitialPicker());
        }
        newChildPolicies.put(entry.getKey(), 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, PolicySelection> adjChildPolicies = new HashMap<>(config.childPolicies);
      for (Entry<String, PolicySelection> 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;
        handleNameResolutionError(state, error);
      }
    }
    if (gotoTransientFailure) {
      getHelper().updateBalancingState(TRANSIENT_FAILURE, getErrorPicker(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, LoadBalancerProvider policyProvider,
        Object childConfig, SubchannelPicker initialPicker) {
      super(key, policyProvider, childConfig, initialPicker);
    }

    @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, PolicySelection> childPolicies = new HashMap<>(config.childPolicies);
          PolicySelection 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();
        }
      }
    }
  }
}

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.LoadBalancerProvider;
27	import io.grpc.Status;
28	import io.grpc.SynchronizationContext;
29	import io.grpc.SynchronizationContext.ScheduledHandle;
30	import io.grpc.internal.ServiceConfigUtil.PolicySelection;
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	Object childConfig) {
75	return resolvedAddresses.toBuilder().setLoadBalancingPolicyConfig(childConfig).build();	1✔
76	}
77
78	@Override
79	protected Map<Object, ChildLbState> createChildLbMap(ResolvedAddresses resolvedAddresses) {
80	ClusterManagerConfig config = (ClusterManagerConfig)	1✔
81	resolvedAddresses.getLoadBalancingPolicyConfig();	1✔
82	Map<Object, ChildLbState> newChildPolicies = new HashMap<>();	1✔
83	if (config != null) {	1✔
84	for (Entry<String, PolicySelection> entry : config.childPolicies.entrySet()) {	1✔
85	ChildLbState child = getChildLbState(entry.getKey());	1✔
86	if (child == null) {	1✔
87	child = new ClusterManagerLbState(entry.getKey(),	1✔
88	entry.getValue().getProvider(), entry.getValue().getConfig(), getInitialPicker());	1✔
89	}
90	newChildPolicies.put(entry.getKey(), child);	1✔
91	}	1✔
92	}
93	logger.log(	1✔
94	XdsLogLevel.INFO,
95	"Received cluster_manager lb config: child names={0}", newChildPolicies.keySet());	1✔
96	return newChildPolicies;	1✔
97	}
98
99	/**
100	* This is like the parent except that it doesn't shutdown the removed children since we want that
101	* to be done by the timer.
102	*/
103	@Override
104	public Status acceptResolvedAddresses(ResolvedAddresses resolvedAddresses) {
105	if (lastResolvedAddresses != null) {	1✔
106	// Handle deactivated children
107	ClusterManagerConfig config = (ClusterManagerConfig)	1✔
108	resolvedAddresses.getLoadBalancingPolicyConfig();	1✔
109	ClusterManagerConfig lastConfig = (ClusterManagerConfig)	1✔
110	lastResolvedAddresses.getLoadBalancingPolicyConfig();	1✔
111	Map<String, PolicySelection> adjChildPolicies = new HashMap<>(config.childPolicies);	1✔
112	for (Entry<String, PolicySelection> entry : lastConfig.childPolicies.entrySet()) {	1✔
113	ClusterManagerLbState state = (ClusterManagerLbState) getChildLbState(entry.getKey());	1✔
114	if (adjChildPolicies.containsKey(entry.getKey())) {	1✔
115	if (state.deletionTimer != null) {	1✔
116	state.reactivateChild();	1✔
117	}
118	} else if (state != null) {	1✔
119	adjChildPolicies.put(entry.getKey(), entry.getValue());	1✔
120	if (state.deletionTimer == null) {	1✔
121	state.deactivateChild();	1✔
122	}
123	}
124	}	1✔
125	config = new ClusterManagerConfig(adjChildPolicies);	1✔
126	resolvedAddresses =	1✔
127	resolvedAddresses.toBuilder().setLoadBalancingPolicyConfig(config).build();	1✔
128	}
129	lastResolvedAddresses = resolvedAddresses;	1✔
130	return super.acceptResolvedAddresses(resolvedAddresses);	1✔
131	}
132
133	/**
134	* Using the state of all children will calculate the current connectivity state,
135	* update currentConnectivityState, generate a picker and then call
136	* {@link Helper#updateBalancingState(ConnectivityState, SubchannelPicker)}.
137	*/
138	@Override
139	protected void updateOverallBalancingState() {
140	ConnectivityState overallState = null;	1✔
141	final Map<Object, SubchannelPicker> childPickers = new HashMap<>();	1✔
142	for (ChildLbState childLbState : getChildLbStates()) {	1✔
143	if (((ClusterManagerLbState) childLbState).deletionTimer != null) {	1✔
144	continue;	1✔
145	}
146	childPickers.put(childLbState.getKey(), childLbState.getCurrentPicker());	1✔
147	overallState = aggregateState(overallState, childLbState.getCurrentState());	1✔
148	}	1✔
149
150	if (overallState != null) {	1✔
151	getHelper().updateBalancingState(overallState, getSubchannelPicker(childPickers));	1✔
152	currentConnectivityState = overallState;	1✔
153	}
154	}	1✔
155
156	protected SubchannelPicker getSubchannelPicker(Map<Object, SubchannelPicker> childPickers) {
157	return new SubchannelPicker() {	1✔
158	@Override
159	public PickResult pickSubchannel(PickSubchannelArgs args) {
160	String clusterName =	1✔
161	args.getCallOptions().getOption(XdsNameResolver.CLUSTER_SELECTION_KEY);	1✔
162	SubchannelPicker childPicker = childPickers.get(clusterName);	1✔
163	if (childPicker == null) {	1✔
164	return	1✔
165	PickResult.withError(	1✔
166	Status.UNAVAILABLE.withDescription("CDS encountered error: unable to find "	1✔
167	+ "available subchannel for cluster " + clusterName));
168	}
169	return childPicker.pickSubchannel(args);	1✔
170	}
171
172	@Override
173	public String toString() {
174	return MoreObjects.toStringHelper(this).add("pickers", childPickers).toString();	×
175	}
176	};
177	}
178
179	@Override
180	public void handleNameResolutionError(Status error) {
181	logger.log(XdsLogLevel.WARNING, "Received name resolution error: {0}", error);	1✔
182	boolean gotoTransientFailure = true;	1✔
183	for (ChildLbState state : getChildLbStates()) {	1✔
184	if (((ClusterManagerLbState) state).deletionTimer == null) {	1✔
185	gotoTransientFailure = false;	1✔
186	handleNameResolutionError(state, error);	1✔
187	}
188	}	1✔
189	if (gotoTransientFailure) {	1✔
190	getHelper().updateBalancingState(TRANSIENT_FAILURE, getErrorPicker(error));	1✔
191	}
192	}	1✔
193
194	/**
195	* This differs from the base class in the use of the deletion timer. When it is deactivated,
196	* rather than immediately calling shutdown it starts a timer. If shutdown or reactivate
197	* are called before the timer fires, the timer is canceled. Otherwise, time timer calls shutdown
198	* and removes the child from the petiole policy when it is triggered.
199	*/
200	private class ClusterManagerLbState extends ChildLbState {	1✔
201	@Nullable
202	ScheduledHandle deletionTimer;
203
204	public ClusterManagerLbState(Object key, LoadBalancerProvider policyProvider,
205	Object childConfig, SubchannelPicker initialPicker) {	1✔
206	super(key, policyProvider, childConfig, initialPicker);	1✔
207	}	1✔
208
209	@Override
210	protected ChildLbStateHelper createChildHelper() {
211	return new ClusterManagerChildHelper();	1✔
212	}
213
214	@Override
215	protected void shutdown() {
216	if (deletionTimer != null) {	1✔
217	deletionTimer.cancel();	1✔
218	deletionTimer = null;	1✔
219	}
220	super.shutdown();	1✔
221	}	1✔
222
223	void reactivateChild() {
224	assert deletionTimer != null;	1✔
225	deletionTimer.cancel();	1✔
226	deletionTimer = null;	1✔
227	logger.log(XdsLogLevel.DEBUG, "Child balancer {0} reactivated", getKey());	1✔
228	}	1✔
229
230	void deactivateChild() {
231	assert deletionTimer == null;	1✔
232
233	class DeletionTask implements Runnable {	1✔
234
235	@Override
236	public void run() {
237	ClusterManagerConfig config = (ClusterManagerConfig)	1✔
238	lastResolvedAddresses.getLoadBalancingPolicyConfig();	1✔
239	Map<String, PolicySelection> childPolicies = new HashMap<>(config.childPolicies);	1✔
240	PolicySelection removed = childPolicies.remove(getKey());	1✔
241	assert removed != null;	1✔
242	config = new ClusterManagerConfig(childPolicies);	1✔
243	lastResolvedAddresses =	1✔
244	lastResolvedAddresses.toBuilder().setLoadBalancingPolicyConfig(config).build();	1✔
245	acceptResolvedAddresses(lastResolvedAddresses);	1✔
246	}	1✔
247	}
248
249	deletionTimer =	1✔
250	syncContext.schedule(	1✔
251	new DeletionTask(),
252	DELAYED_CHILD_DELETION_TIME_MINUTES,
253	TimeUnit.MINUTES,
254	timeService);	1✔
255	logger.log(XdsLogLevel.DEBUG, "Child balancer {0} deactivated", getKey());	1✔
256	}	1✔
257
258	private class ClusterManagerChildHelper extends ChildLbStateHelper {	1✔
259	@Override
260	public void updateBalancingState(final ConnectivityState newState,
261	final SubchannelPicker newPicker) {
262	// If we are already in the process of resolving addresses, the overall balancing state
263	// will be updated at the end of it, and we don't need to trigger that update here.
264	if (getChildLbState(getKey()) == null) {	1✔
265	return;	1✔
266	}
267
268	// Subchannel picker and state are saved, but will only be propagated to the channel
269	// when the child instance exits deactivated state.
270	setCurrentState(newState);	1✔
271	setCurrentPicker(newPicker);	1✔
272	if (deletionTimer == null && !resolvingAddresses) {	1✔
273	updateOverallBalancingState();	1✔
274	}
275	}	1✔
276	}
277	}
278	}

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