#20365

Committed 27 Jul 2026 06:04AM UTC coverage: 89.199% (+0.07%) from 89.125%

Build # #20365

Build Type

push

github

Committed by

web-flow

Commit Message

core: Implement LB Delay Observability (Proposal A121) (#12807)

This PR implements **Attempt-Level RPC Delay Observability** across the core channel transport, built-in load balancers, xDS policies, and the OpenTelemetry telemetry plugin, aligned with [gRPC Proposal A121](https://github.com/grpc/proposal/pull/556).

Coverage Stats

38276 of 42911 relevant lines covered (89.2%)

0.89 hits per line

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

99.49

/../xds/src/main/java/io/grpc/xds/PriorityLoadBalancer.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.CONNECTING;
import static io.grpc.ConnectivityState.IDLE;
import static io.grpc.ConnectivityState.READY;
import static io.grpc.ConnectivityState.TRANSIENT_FAILURE;

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.internal.GrpcUtil;
import io.grpc.util.ForwardingLoadBalancerHelper;
import io.grpc.util.GracefulSwitchLoadBalancer;
import io.grpc.xds.PriorityLoadBalancerProvider.PriorityLbConfig;
import io.grpc.xds.PriorityLoadBalancerProvider.PriorityLbConfig.PriorityChildConfig;
import io.grpc.xds.client.XdsLogger;
import io.grpc.xds.client.XdsLogger.XdsLogLevel;
import java.util.ArrayList;
import java.util.Collection;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Set;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import javax.annotation.Nullable;

/**
 * Load balancer for priority policy. A <em>priority</em> represents a logical entity within a
 * cluster for load balancing purposes.
 */
final class PriorityLoadBalancer extends LoadBalancer {
  private final Helper helper;
  private final SynchronizationContext syncContext;
  private final ScheduledExecutorService executor;
  private final XdsLogger logger;

  // Includes all active and deactivated children. Mutable. New entries are only added from priority
  // 0 up to the selected priority. An entry is only deleted 15 minutes after its deactivation.
  // Note that calling into a child can cause the child to call back into the LB policy and modify
  // the map.  Therefore copy values before looping over them.
  private final Map<String, ChildLbState> children = new HashMap<>();

  // Following fields are only null initially.
  private ResolvedAddresses resolvedAddresses;
  // List of priority names in order.
  private List<String> priorityNames;
  // Config for each priority.
  private Map<String, PriorityChildConfig> priorityConfigs;
  @Nullable private String currentPriority;
  private ConnectivityState currentConnectivityState;
  private SubchannelPicker currentPicker;
  // Set to true if currently in the process of handling resolved addresses.
  private boolean handlingResolvedAddresses;
  static boolean enablePriorityLbChildPolicyCache =
      GrpcUtil.getFlag("GRPC_EXPERIMENTAL_ENABLE_PRIORITY_LB_CHILD_POLICY_CACHE", false);

  PriorityLoadBalancer(Helper helper) {
    this.helper = checkNotNull(helper, "helper");
    syncContext = helper.getSynchronizationContext();
    executor = helper.getScheduledExecutorService();
    InternalLogId logId = InternalLogId.allocate("priority-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);
    this.resolvedAddresses = resolvedAddresses;
    PriorityLbConfig config = (PriorityLbConfig) resolvedAddresses.getLoadBalancingPolicyConfig();
    checkNotNull(config, "missing priority lb config");
    priorityNames = config.priorities;
    priorityConfigs = config.childConfigs;
    Status status = Status.OK;
    Set<String> prioritySet = new HashSet<>(config.priorities);
    ArrayList<String> childKeys = new ArrayList<>(children.keySet());
    for (String priority : childKeys) {
      if (!prioritySet.contains(priority)) {
        ChildLbState childLbState = children.get(priority);
        if (childLbState != null) {
          if (enablePriorityLbChildPolicyCache) {
            childLbState.deactivate();
          } else {
            childLbState.tearDown();
            children.remove(priority);
          }
        }
      }
    }
    handlingResolvedAddresses = true;
    for (String priority : priorityNames) {
      ChildLbState childLbState = children.get(priority);
      if (childLbState != null) {
        Status newStatus = childLbState.updateResolvedAddresses();
        if (!newStatus.isOk()) {
          status = newStatus;
        }
      }
    }
    handlingResolvedAddresses = false;
    Status newStatus = tryNextPriority();
    if (!newStatus.isOk()) {
      status = newStatus;
    }
    return status;
  }

  @Override
  public void handleNameResolutionError(Status error) {
    logger.log(XdsLogLevel.WARNING, "Received name resolution error: {0}", error);
    boolean gotoTransientFailure = true;
    Collection<ChildLbState> childValues = new ArrayList<>(children.values());
    for (ChildLbState child : childValues) {
      if (priorityNames.contains(child.priority)) {
        child.lb.handleNameResolutionError(error);
        gotoTransientFailure = false;
      }
    }
    if (gotoTransientFailure) {
      updateOverallState(
          null, TRANSIENT_FAILURE, new FixedResultPicker(PickResult.withError(error)));
    }
  }

  @Override
  public void shutdown() {
    logger.log(XdsLogLevel.INFO, "Shutdown");
    Collection<ChildLbState> childValues = new ArrayList<>(children.values());
    for (ChildLbState child : childValues) {
      child.tearDown();
    }
    children.clear();
  }

  private Status tryNextPriority() {
    for (int i = 0; i < priorityNames.size(); i++) {
      String priority = priorityNames.get(i);
      if (!children.containsKey(priority)) {
        ChildLbState child =
            new ChildLbState(priority, priorityConfigs.get(priority).ignoreReresolution);
        children.put(priority, child);
        // Child is created in CONNECTING with pending failOverTimer
        updateOverallState(priority, child.connectivityState, child.picker);
        // Calling the child's updateResolvedAddresses() can result in tryNextPriority() being
        // called recursively. We need to be sure to be done with processing here before it is
        // called.
        return child.updateResolvedAddresses(); // Give priority i time to connect.
      }
      ChildLbState child = children.get(priority);
      child.reactivate();
      if (child.connectivityState.equals(READY) || child.connectivityState.equals(IDLE)) {
        logger.log(XdsLogLevel.DEBUG, "Shifted to priority {0}", priority);
        updateOverallState(priority, child.connectivityState, child.picker);
        for (int j = i + 1; j < priorityNames.size(); j++) {
          String p = priorityNames.get(j);
          if (children.containsKey(p)) {
            children.get(p).deactivate();
          }
        }
        return Status.OK;
      }
      if (child.failOverTimer.isPending()) {
        updateOverallState(priority, child.connectivityState, child.picker);
        return Status.OK; // Give priority i time to connect.
      }
    }
    for (int i = 0; i < priorityNames.size(); i++) {
      String priority = priorityNames.get(i);
      ChildLbState child = children.get(priority);
      if (child.connectivityState.equals(CONNECTING)) {
        updateOverallState(priority, child.connectivityState, child.picker);
        return Status.OK;
      }
    }
    logger.log(XdsLogLevel.DEBUG, "All priority failed");
    String lastPriority = priorityNames.get(priorityNames.size() - 1);
    ChildLbState child = children.get(lastPriority);
    updateOverallState(lastPriority, child.connectivityState, child.picker);
    return Status.OK;
  }

  private void updateOverallState(
      @Nullable String priority, ConnectivityState state, SubchannelPicker picker) {
    if (!Objects.equals(priority, currentPriority) || !state.equals(currentConnectivityState)
        || !picker.equals(currentPicker)) {
      currentPriority = priority;
      currentConnectivityState = state;
      currentPicker = picker;
      helper.updateBalancingState(state, picker);
    }
  }

  private final class ChildLbState {
    final String priority;
    final ChildHelper childHelper;
    final GracefulSwitchLoadBalancer lb;
    // Timer to fail over to the next priority if not connected in 10 sec. Scheduled only once at
    // child initialization.
    ScheduledHandle failOverTimer;
    boolean seenReadyOrIdleSinceTransientFailure = false;
    // Timer to delay shutdown and deletion of the priority. Scheduled whenever the child is
    // deactivated.
    @Nullable ScheduledHandle deletionTimer;
    ConnectivityState connectivityState = CONNECTING;
    SubchannelPicker picker = new FixedResultPicker(
        PickResult.withNoResult("connecting", "priority child state uninitialized"));

    ChildLbState(final String priority, boolean ignoreReresolution) {
      this.priority = priority;
      childHelper = new ChildHelper(ignoreReresolution);
      lb = new GracefulSwitchLoadBalancer(childHelper);
      failOverTimer = syncContext.schedule(new FailOverTask(), 10, TimeUnit.SECONDS, executor);
      logger.log(XdsLogLevel.DEBUG, "Priority created: {0}", priority);
    }

    final class FailOverTask implements Runnable {
      @Override
      public void run() {
        if (deletionTimer != null && deletionTimer.isPending()) {
          // The child is deactivated.
          return;
        }
        logger.log(XdsLogLevel.DEBUG, "Priority {0} failed over to next", priority);
        Status status = tryNextPriority();
        if (!status.isOk()) {
          // A child had a problem with the addresses/config. Request it to be refreshed
          helper.refreshNameResolution();
        }
      }
    }

    /**
     * Called when the child becomes a priority that is or appears before the first READY one in the
     * {@code priorities} list, due to either config update or balancing state update.
     */
    void reactivate() {
      if (deletionTimer != null && deletionTimer.isPending()) {
        deletionTimer.cancel();
        logger.log(XdsLogLevel.DEBUG, "Priority reactivated: {0}", priority);
      }
    }

    /**
     * Called when either the child is removed by config update, or a higher priority becomes READY.
     */
    void deactivate() {
      if (deletionTimer != null && deletionTimer.isPending()) {
        return;
      }

      class DeletionTask implements Runnable {
        @Override
        public void run() {
          tearDown();
          children.remove(priority);
        }
      }

      deletionTimer = syncContext.schedule(new DeletionTask(), 15, TimeUnit.MINUTES, executor);
      logger.log(XdsLogLevel.DEBUG, "Priority deactivated: {0}", priority);
    }

    void tearDown() {
      if (failOverTimer.isPending()) {
        failOverTimer.cancel();
      }
      if (deletionTimer != null && deletionTimer.isPending()) {
        deletionTimer.cancel();
      }
      lb.shutdown();
      logger.log(XdsLogLevel.DEBUG, "Priority deleted: {0}", priority);
    }

    /**
     * Called either when the child is just created and in this case updated with the cached {@code
     * resolvedAddresses}, or when priority lb receives a new resolved addresses while the child
     * already exists.
     */
    Status updateResolvedAddresses() {
      PriorityLbConfig config =
          (PriorityLbConfig) resolvedAddresses.getLoadBalancingPolicyConfig();
      return lb.acceptResolvedAddresses(
          resolvedAddresses.toBuilder()
              .setAddresses(AddressFilter.filter(resolvedAddresses.getAddresses(), priority))
              .setLoadBalancingPolicyConfig(config.childConfigs.get(priority).childConfig)
              .build());
    }

    final class ChildHelper extends ForwardingLoadBalancerHelper {
      private final boolean ignoreReresolution;

      ChildHelper(boolean ignoreReresolution) {
        this.ignoreReresolution = ignoreReresolution;
      }

      @Override
      public void refreshNameResolution() {
        if (!ignoreReresolution) {
          delegate().refreshNameResolution();
        }
      }

      @Override
      public void updateBalancingState(final ConnectivityState newState,
          final SubchannelPicker newPicker) {
        if (!children.containsKey(priority)) {
          return;
        }
        ConnectivityState oldState = connectivityState;
        connectivityState = newState;
        if (newState == CONNECTING || newState == IDLE) {
          picker = new PriorityPicker(newPicker, priority);
        } else {
          picker = newPicker;
        }

        if (deletionTimer != null && deletionTimer.isPending()) {
          return;
        }
        if (newState.equals(CONNECTING) && !oldState.equals(newState)) {
          if (!failOverTimer.isPending() && seenReadyOrIdleSinceTransientFailure) {
            failOverTimer = syncContext.schedule(new FailOverTask(), 10, TimeUnit.SECONDS,
                executor);
          }
        } else if (newState.equals(READY) || newState.equals(IDLE)) {
          seenReadyOrIdleSinceTransientFailure = true;
          failOverTimer.cancel();
        } else if (newState.equals(TRANSIENT_FAILURE)) {
          seenReadyOrIdleSinceTransientFailure = false;
          failOverTimer.cancel();
        }

        // If we are currently handling newly resolved addresses, let's not try to reconfigure as
        // the address handling process will take care of that to provide an atomic config update.
        if (!handlingResolvedAddresses) {
          Status status = tryNextPriority();
          if (!status.isOk()) {
            // A child had a problem with the addresses/config. Request it to be refreshed
            helper.refreshNameResolution();
          }
        }
      }

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

  private static final class PriorityPicker extends SubchannelPicker {
    private final SubchannelPicker delegate;
    private final String priority;

    PriorityPicker(SubchannelPicker delegate, String priority) {
      this.delegate = checkNotNull(delegate, "delegate");
      this.priority = checkNotNull(priority, "priority");
    }

    @Override
    public PickResult pickSubchannel(PickSubchannelArgs args) {
      PickResult childResult = delegate.pickSubchannel(args);
      if (!childResult.hasResult() && childResult.getDelayType() != null) {
        String childReason = childResult.getDelayReason();
        String composedType = priority + ":" + childResult.getDelayType();
        String reason = "waiting on priority group " + priority + " ("
            + (childReason != null ? childReason : "connecting") + ")";
        return PickResult.withNoResult(composedType, reason);
      }
      return childResult;
    }

    @Override
    public boolean equals(Object o) {
      if (this == o) {
        return true;
      }
      if (o == null || getClass() != o.getClass()) {
        return false;
      }
      PriorityPicker that = (PriorityPicker) o;
      return delegate.equals(that.delegate) && priority.equals(that.priority);
    }

    @Override
    public int hashCode() {
      return Objects.hash(delegate, priority);
    }
  }
}

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.CONNECTING;
21	import static io.grpc.ConnectivityState.IDLE;
22	import static io.grpc.ConnectivityState.READY;
23	import static io.grpc.ConnectivityState.TRANSIENT_FAILURE;
24
25	import io.grpc.ConnectivityState;
26	import io.grpc.InternalLogId;
27	import io.grpc.LoadBalancer;
28	import io.grpc.Status;
29	import io.grpc.SynchronizationContext;
30	import io.grpc.SynchronizationContext.ScheduledHandle;
31	import io.grpc.internal.GrpcUtil;
32	import io.grpc.util.ForwardingLoadBalancerHelper;
33	import io.grpc.util.GracefulSwitchLoadBalancer;
34	import io.grpc.xds.PriorityLoadBalancerProvider.PriorityLbConfig;
35	import io.grpc.xds.PriorityLoadBalancerProvider.PriorityLbConfig.PriorityChildConfig;
36	import io.grpc.xds.client.XdsLogger;
37	import io.grpc.xds.client.XdsLogger.XdsLogLevel;
38	import java.util.ArrayList;
39	import java.util.Collection;
40	import java.util.HashMap;
41	import java.util.HashSet;
42	import java.util.List;
43	import java.util.Map;
44	import java.util.Objects;
45	import java.util.Set;
46	import java.util.concurrent.ScheduledExecutorService;
47	import java.util.concurrent.TimeUnit;
48	import javax.annotation.Nullable;
49
50	/**
51	* Load balancer for priority policy. A <em>priority</em> represents a logical entity within a
52	* cluster for load balancing purposes.
53	*/
54	final class PriorityLoadBalancer extends LoadBalancer {
55	private final Helper helper;
56	private final SynchronizationContext syncContext;
57	private final ScheduledExecutorService executor;
58	private final XdsLogger logger;
59
60	// Includes all active and deactivated children. Mutable. New entries are only added from priority
61	// 0 up to the selected priority. An entry is only deleted 15 minutes after its deactivation.
62	// Note that calling into a child can cause the child to call back into the LB policy and modify
63	// the map. Therefore copy values before looping over them.
64	private final Map<String, ChildLbState> children = new HashMap<>();	1✔
65
66	// Following fields are only null initially.
67	private ResolvedAddresses resolvedAddresses;
68	// List of priority names in order.
69	private List<String> priorityNames;
70	// Config for each priority.
71	private Map<String, PriorityChildConfig> priorityConfigs;
72	@Nullable private String currentPriority;
73	private ConnectivityState currentConnectivityState;
74	private SubchannelPicker currentPicker;
75	// Set to true if currently in the process of handling resolved addresses.
76	private boolean handlingResolvedAddresses;
77	static boolean enablePriorityLbChildPolicyCache =	1✔
78	GrpcUtil.getFlag("GRPC_EXPERIMENTAL_ENABLE_PRIORITY_LB_CHILD_POLICY_CACHE", false);	1✔
79
80	PriorityLoadBalancer(Helper helper) {	1✔
81	this.helper = checkNotNull(helper, "helper");	1✔
82	syncContext = helper.getSynchronizationContext();	1✔
83	executor = helper.getScheduledExecutorService();	1✔
84	InternalLogId logId = InternalLogId.allocate("priority-lb", helper.getAuthority());	1✔
85	logger = XdsLogger.withLogId(logId);	1✔
86	logger.log(XdsLogLevel.INFO, "Created");	1✔
87	}	1✔
88
89	@Override
90	public Status acceptResolvedAddresses(ResolvedAddresses resolvedAddresses) {
91	logger.log(XdsLogLevel.DEBUG, "Received resolution result: {0}", resolvedAddresses);	1✔
92	this.resolvedAddresses = resolvedAddresses;	1✔
93	PriorityLbConfig config = (PriorityLbConfig) resolvedAddresses.getLoadBalancingPolicyConfig();	1✔
94	checkNotNull(config, "missing priority lb config");	1✔
95	priorityNames = config.priorities;	1✔
96	priorityConfigs = config.childConfigs;	1✔
97	Status status = Status.OK;	1✔
98	Set<String> prioritySet = new HashSet<>(config.priorities);	1✔
99	ArrayList<String> childKeys = new ArrayList<>(children.keySet());	1✔
100	for (String priority : childKeys) {	1✔
101	if (!prioritySet.contains(priority)) {	1✔
102	ChildLbState childLbState = children.get(priority);	1✔
103	if (childLbState != null) {	1✔
104	if (enablePriorityLbChildPolicyCache) {	1✔
105	childLbState.deactivate();	1✔
106	} else {
107	childLbState.tearDown();	1✔
108	children.remove(priority);	1✔
109	}
110	}
111	}
112	}	1✔
113	handlingResolvedAddresses = true;	1✔
114	for (String priority : priorityNames) {	1✔
115	ChildLbState childLbState = children.get(priority);	1✔
116	if (childLbState != null) {	1✔
117	Status newStatus = childLbState.updateResolvedAddresses();	1✔
118	if (!newStatus.isOk()) {	1✔
119	status = newStatus;	1✔
120	}
121	}
122	}	1✔
123	handlingResolvedAddresses = false;	1✔
124	Status newStatus = tryNextPriority();	1✔
125	if (!newStatus.isOk()) {	1✔
126	status = newStatus;	1✔
127	}
128	return status;	1✔
129	}
130
131	@Override
132	public void handleNameResolutionError(Status error) {
133	logger.log(XdsLogLevel.WARNING, "Received name resolution error: {0}", error);	1✔
134	boolean gotoTransientFailure = true;	1✔
135	Collection<ChildLbState> childValues = new ArrayList<>(children.values());	1✔
136	for (ChildLbState child : childValues) {	1✔
137	if (priorityNames.contains(child.priority)) {	1✔
138	child.lb.handleNameResolutionError(error);	1✔
139	gotoTransientFailure = false;	1✔
140	}
141	}	1✔
142	if (gotoTransientFailure) {	1✔
143	updateOverallState(	1✔
144	null, TRANSIENT_FAILURE, new FixedResultPicker(PickResult.withError(error)));	1✔
145	}
146	}	1✔
147
148	@Override
149	public void shutdown() {
150	logger.log(XdsLogLevel.INFO, "Shutdown");	1✔
151	Collection<ChildLbState> childValues = new ArrayList<>(children.values());	1✔
152	for (ChildLbState child : childValues) {	1✔
153	child.tearDown();	1✔
154	}	1✔
155	children.clear();	1✔
156	}	1✔
157
158	private Status tryNextPriority() {
159	for (int i = 0; i < priorityNames.size(); i++) {	1✔
160	String priority = priorityNames.get(i);	1✔
161	if (!children.containsKey(priority)) {	1✔
162	ChildLbState child =	1✔
163	new ChildLbState(priority, priorityConfigs.get(priority).ignoreReresolution);	1✔
164	children.put(priority, child);	1✔
165	// Child is created in CONNECTING with pending failOverTimer
166	updateOverallState(priority, child.connectivityState, child.picker);	1✔
167	// Calling the child's updateResolvedAddresses() can result in tryNextPriority() being
168	// called recursively. We need to be sure to be done with processing here before it is
169	// called.
170	return child.updateResolvedAddresses(); // Give priority i time to connect.	1✔
171	}
172	ChildLbState child = children.get(priority);	1✔
173	child.reactivate();	1✔
174	if (child.connectivityState.equals(READY) \|\| child.connectivityState.equals(IDLE)) {	1✔
175	logger.log(XdsLogLevel.DEBUG, "Shifted to priority {0}", priority);	1✔
176	updateOverallState(priority, child.connectivityState, child.picker);	1✔
177	for (int j = i + 1; j < priorityNames.size(); j++) {	1✔
178	String p = priorityNames.get(j);	1✔
179	if (children.containsKey(p)) {	1✔
180	children.get(p).deactivate();	1✔
181	}
182	}
183	return Status.OK;	1✔
184	}
185	if (child.failOverTimer.isPending()) {	1✔
186	updateOverallState(priority, child.connectivityState, child.picker);	1✔
187	return Status.OK; // Give priority i time to connect.	1✔
188	}
189	}
190	for (int i = 0; i < priorityNames.size(); i++) {	1✔
191	String priority = priorityNames.get(i);	1✔
192	ChildLbState child = children.get(priority);	1✔
193	if (child.connectivityState.equals(CONNECTING)) {	1✔
194	updateOverallState(priority, child.connectivityState, child.picker);	1✔
195	return Status.OK;	1✔
196	}
197	}
198	logger.log(XdsLogLevel.DEBUG, "All priority failed");	1✔
199	String lastPriority = priorityNames.get(priorityNames.size() - 1);	1✔
200	ChildLbState child = children.get(lastPriority);	1✔
201	updateOverallState(lastPriority, child.connectivityState, child.picker);	1✔
202	return Status.OK;	1✔
203	}
204
205	private void updateOverallState(
206	@Nullable String priority, ConnectivityState state, SubchannelPicker picker) {
207	if (!Objects.equals(priority, currentPriority) \|\| !state.equals(currentConnectivityState)	1✔
208	\|\| !picker.equals(currentPicker)) {	1✔
209	currentPriority = priority;	1✔
210	currentConnectivityState = state;	1✔
211	currentPicker = picker;	1✔
212	helper.updateBalancingState(state, picker);	1✔
213	}
214	}	1✔
215
216	private final class ChildLbState {
217	final String priority;
218	final ChildHelper childHelper;
219	final GracefulSwitchLoadBalancer lb;
220	// Timer to fail over to the next priority if not connected in 10 sec. Scheduled only once at
221	// child initialization.
222	ScheduledHandle failOverTimer;
223	boolean seenReadyOrIdleSinceTransientFailure = false;	1✔
224	// Timer to delay shutdown and deletion of the priority. Scheduled whenever the child is
225	// deactivated.
226	@Nullable ScheduledHandle deletionTimer;
227	ConnectivityState connectivityState = CONNECTING;	1✔
228	SubchannelPicker picker = new FixedResultPicker(	1✔
229	PickResult.withNoResult("connecting", "priority child state uninitialized"));	1✔
230
231	ChildLbState(final String priority, boolean ignoreReresolution) {	1✔
232	this.priority = priority;	1✔
233	childHelper = new ChildHelper(ignoreReresolution);	1✔
234	lb = new GracefulSwitchLoadBalancer(childHelper);	1✔
235	failOverTimer = syncContext.schedule(new FailOverTask(), 10, TimeUnit.SECONDS, executor);	1✔
236	logger.log(XdsLogLevel.DEBUG, "Priority created: {0}", priority);	1✔
237	}	1✔
238
239	final class FailOverTask implements Runnable {	1✔
240	@Override
241	public void run() {
242	if (deletionTimer != null && deletionTimer.isPending()) {	1✔
243	// The child is deactivated.
244	return;	1✔
245	}
246	logger.log(XdsLogLevel.DEBUG, "Priority {0} failed over to next", priority);	1✔
247	Status status = tryNextPriority();	1✔
248	if (!status.isOk()) {	1✔
249	// A child had a problem with the addresses/config. Request it to be refreshed
250	helper.refreshNameResolution();	1✔
251	}
252	}	1✔
253	}
254
255	/**
256	* Called when the child becomes a priority that is or appears before the first READY one in the
257	* {@code priorities} list, due to either config update or balancing state update.
258	*/
259	void reactivate() {
260	if (deletionTimer != null && deletionTimer.isPending()) {	1✔
261	deletionTimer.cancel();	1✔
262	logger.log(XdsLogLevel.DEBUG, "Priority reactivated: {0}", priority);	1✔
263	}
264	}	1✔
265
266	/**
267	* Called when either the child is removed by config update, or a higher priority becomes READY.
268	*/
269	void deactivate() {
270	if (deletionTimer != null && deletionTimer.isPending()) {	1✔
271	return;	1✔
272	}
273
274	class DeletionTask implements Runnable {	1✔
275	@Override
276	public void run() {
277	tearDown();	1✔
278	children.remove(priority);	1✔
279	}	1✔
280	}
281
282	deletionTimer = syncContext.schedule(new DeletionTask(), 15, TimeUnit.MINUTES, executor);	1✔
283	logger.log(XdsLogLevel.DEBUG, "Priority deactivated: {0}", priority);	1✔
284	}	1✔
285
286	void tearDown() {
287	if (failOverTimer.isPending()) {	1✔
288	failOverTimer.cancel();	1✔
289	}
290	if (deletionTimer != null && deletionTimer.isPending()) {	1✔
291	deletionTimer.cancel();	1✔
292	}
293	lb.shutdown();	1✔
294	logger.log(XdsLogLevel.DEBUG, "Priority deleted: {0}", priority);	1✔
295	}	1✔
296
297	/**
298	* Called either when the child is just created and in this case updated with the cached {@code
299	* resolvedAddresses}, or when priority lb receives a new resolved addresses while the child
300	* already exists.
301	*/
302	Status updateResolvedAddresses() {
303	PriorityLbConfig config =	1✔
304	(PriorityLbConfig) resolvedAddresses.getLoadBalancingPolicyConfig();	1✔
305	return lb.acceptResolvedAddresses(	1✔
306	resolvedAddresses.toBuilder()	1✔
307	.setAddresses(AddressFilter.filter(resolvedAddresses.getAddresses(), priority))	1✔
308	.setLoadBalancingPolicyConfig(config.childConfigs.get(priority).childConfig)	1✔
309	.build());	1✔
310	}
311
312	final class ChildHelper extends ForwardingLoadBalancerHelper {
313	private final boolean ignoreReresolution;
314
315	ChildHelper(boolean ignoreReresolution) {	1✔
316	this.ignoreReresolution = ignoreReresolution;	1✔
317	}	1✔
318
319	@Override
320	public void refreshNameResolution() {
321	if (!ignoreReresolution) {	1✔
322	delegate().refreshNameResolution();	1✔
323	}
324	}	1✔
325
326	@Override
327	public void updateBalancingState(final ConnectivityState newState,
328	final SubchannelPicker newPicker) {
329	if (!children.containsKey(priority)) {	1✔
330	return;	1✔
331	}
332	ConnectivityState oldState = connectivityState;	1✔
333	connectivityState = newState;	1✔
334	if (newState == CONNECTING \|\| newState == IDLE) {	1✔
335	picker = new PriorityPicker(newPicker, priority);	1✔
336	} else {
337	picker = newPicker;	1✔
338	}
339
340	if (deletionTimer != null && deletionTimer.isPending()) {	1✔
341	return;	1✔
342	}
343	if (newState.equals(CONNECTING) && !oldState.equals(newState)) {	1✔
344	if (!failOverTimer.isPending() && seenReadyOrIdleSinceTransientFailure) {	1✔
345	failOverTimer = syncContext.schedule(new FailOverTask(), 10, TimeUnit.SECONDS,	1✔
346	executor);	1✔
347	}
348	} else if (newState.equals(READY) \|\| newState.equals(IDLE)) {	1✔
349	seenReadyOrIdleSinceTransientFailure = true;	1✔
350	failOverTimer.cancel();	1✔
351	} else if (newState.equals(TRANSIENT_FAILURE)) {	1✔
352	seenReadyOrIdleSinceTransientFailure = false;	1✔
353	failOverTimer.cancel();	1✔
354	}
355
356	// If we are currently handling newly resolved addresses, let's not try to reconfigure as
357	// the address handling process will take care of that to provide an atomic config update.
358	if (!handlingResolvedAddresses) {	1✔
359	Status status = tryNextPriority();	1✔
360	if (!status.isOk()) {	1✔
361	// A child had a problem with the addresses/config. Request it to be refreshed
362	helper.refreshNameResolution();	1✔
363	}
364	}
365	}	1✔
366
367	@Override
368	protected Helper delegate() {
369	return helper;	1✔
370	}
371	}
372	}
373
374	private static final class PriorityPicker extends SubchannelPicker {
375	private final SubchannelPicker delegate;
376	private final String priority;
377
378	PriorityPicker(SubchannelPicker delegate, String priority) {	1✔
379	this.delegate = checkNotNull(delegate, "delegate");	1✔
380	this.priority = checkNotNull(priority, "priority");	1✔
381	}	1✔
382
383	@Override
384	public PickResult pickSubchannel(PickSubchannelArgs args) {
385	PickResult childResult = delegate.pickSubchannel(args);	1✔
386	if (!childResult.hasResult() && childResult.getDelayType() != null) {	1✔
387	String childReason = childResult.getDelayReason();	1✔
388	String composedType = priority + ":" + childResult.getDelayType();	1✔
389	String reason = "waiting on priority group " + priority + " ("	1✔
390	+ (childReason != null ? childReason : "connecting") + ")";	1✔
391	return PickResult.withNoResult(composedType, reason);	1✔
392	}
393	return childResult;	1✔
394	}
395
396	@Override
397	public boolean equals(Object o) {
398	if (this == o) {	1✔
399	return true;	1✔
400	}
401	if (o == null \|\| getClass() != o.getClass()) {	1✔
402	return false;	1✔
403	}
404	PriorityPicker that = (PriorityPicker) o;	1✔
405	return delegate.equals(that.delegate) && priority.equals(that.priority);	1✔
406	}
407
408	@Override
409	public int hashCode() {
410	return Objects.hash(delegate, priority);	×
411	}
412	}
413	}

grpc / grpc-java / #20365

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