#20057

Committed 06 Nov 2025 10:27AM UTC coverage: 88.528% (-0.01%) from 88.542%

Build # #20057

Build Type

push

github

Committed by

web-flow

Commit Message

rls: Control plane channel monitor state and back off handling (#12460)

At the end of back off time, instead of firing a Rls RPC, just update the RLS picker, and RLS connectivity state change from TRANSIENT_FAILURE to READY deactivate all active backoffs.

Run Details

35044 of 39585 relevant lines covered (88.53%)

0.89 hits per line

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

79.12

/../rls/src/main/java/io/grpc/rls/LbPolicyConfiguration.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.rls;

import static com.google.common.base.Preconditions.checkArgument;
import static com.google.common.base.Preconditions.checkNotNull;
import static com.google.common.base.Preconditions.checkState;

import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.MoreObjects;
import io.grpc.ChannelLogger.ChannelLogLevel;
import io.grpc.ConnectivityState;
import io.grpc.LoadBalancer;
import io.grpc.LoadBalancer.Helper;
import io.grpc.LoadBalancer.Subchannel;
import io.grpc.LoadBalancer.SubchannelPicker;
import io.grpc.LoadBalancerProvider;
import io.grpc.LoadBalancerRegistry;
import io.grpc.NameResolver.ConfigOrError;
import io.grpc.Status;
import io.grpc.internal.ObjectPool;
import io.grpc.rls.ChildLoadBalancerHelper.ChildLoadBalancerHelperProvider;
import io.grpc.rls.RlsProtoData.RouteLookupConfig;
import io.grpc.util.ForwardingLoadBalancerHelper;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import javax.annotation.Nullable;

/** Configuration for RLS load balancing policy. */
final class LbPolicyConfiguration {

  private final RouteLookupConfig routeLookupConfig;
  @Nullable
  private final Map<String, ?> routeLookupChannelServiceConfig;
  private final ChildLoadBalancingPolicy policy;

  LbPolicyConfiguration(
      RouteLookupConfig routeLookupConfig, @Nullable Map<String, ?> routeLookupChannelServiceConfig,
      ChildLoadBalancingPolicy policy) {
    this.routeLookupConfig = checkNotNull(routeLookupConfig, "routeLookupConfig");
    this.routeLookupChannelServiceConfig = routeLookupChannelServiceConfig;
    this.policy = checkNotNull(policy, "policy");
  }

  RouteLookupConfig getRouteLookupConfig() {
    return routeLookupConfig;
  }

  @Nullable
  Map<String, ?> getRouteLookupChannelServiceConfig() {
    return routeLookupChannelServiceConfig;
  }

  ChildLoadBalancingPolicy getLoadBalancingPolicy() {
    return policy;
  }

  @Override
  public boolean equals(Object o) {
    if (this == o) {
      return true;
    }
    if (o == null || getClass() != o.getClass()) {
      return false;
    }
    LbPolicyConfiguration that = (LbPolicyConfiguration) o;
    return Objects.equals(routeLookupConfig, that.routeLookupConfig)
        && Objects.equals(routeLookupChannelServiceConfig, that.routeLookupChannelServiceConfig)
        && Objects.equals(policy, that.policy);
  }

  @Override
  public int hashCode() {
    return Objects.hash(routeLookupConfig, routeLookupChannelServiceConfig, policy);
  }

  @Override
  public String toString() {
    return MoreObjects.toStringHelper(this)
        .add("routeLookupConfig", routeLookupConfig)
        .add("routeLookupChannelServiceConfig", routeLookupChannelServiceConfig)
        .add("policy", policy)
        .toString();
  }

  /** ChildLoadBalancingPolicy is an elected child policy to delegate requests. */
  static final class ChildLoadBalancingPolicy {

    private final Map<String, Object> effectiveRawChildPolicy;
    private final LoadBalancerProvider effectiveLbProvider;
    private final String targetFieldName;

    @VisibleForTesting
    ChildLoadBalancingPolicy(
        String targetFieldName,
        Map<String, Object> effectiveRawChildPolicy,
        LoadBalancerProvider effectiveLbProvider) {
      checkArgument(
          targetFieldName != null && !targetFieldName.isEmpty(),
          "targetFieldName cannot be empty or null");
      this.targetFieldName = targetFieldName;
      this.effectiveRawChildPolicy =
          checkNotNull(effectiveRawChildPolicy, "effectiveRawChildPolicy");
      this.effectiveLbProvider = checkNotNull(effectiveLbProvider, "effectiveLbProvider");
    }

    /** Creates ChildLoadBalancingPolicy. */
    @SuppressWarnings("unchecked")
    static ChildLoadBalancingPolicy create(
        String childPolicyConfigTargetFieldName, List<Map<String, ?>> childPolicies)
        throws InvalidChildPolicyConfigException {
      Map<String, Object> effectiveChildPolicy = null;
      LoadBalancerProvider effectiveLbProvider = null;
      List<String> policyTried = new ArrayList<>();

      LoadBalancerRegistry lbRegistry = LoadBalancerRegistry.getDefaultRegistry();
      for (Map<String, ?> childPolicy : childPolicies) {
        if (childPolicy.isEmpty()) {
          continue;
        }
        if (childPolicy.size() != 1) {
          throw
              new InvalidChildPolicyConfigException(
                  "childPolicy should have exactly one loadbalancing policy");
        }
        String policyName = childPolicy.keySet().iterator().next();
        LoadBalancerProvider provider = lbRegistry.getProvider(policyName);
        if (provider != null) {
          effectiveLbProvider = provider;
          effectiveChildPolicy = Collections.unmodifiableMap(childPolicy);
          break;
        }
        policyTried.add(policyName);
      }
      if (effectiveChildPolicy == null) {
        throw
            new InvalidChildPolicyConfigException(
                String.format("no valid childPolicy found, policy tried: %s", policyTried));
      }
      return
          new ChildLoadBalancingPolicy(
              childPolicyConfigTargetFieldName,
              (Map<String, Object>) effectiveChildPolicy.values().iterator().next(),
              effectiveLbProvider);
    }

    /** Creates a child load balancer config for given target from elected raw child policy. */
    Map<String, ?> getEffectiveChildPolicy(String target) {
      Map<String, Object> childPolicy = new HashMap<>(effectiveRawChildPolicy);
      childPolicy.put(targetFieldName, target);
      return childPolicy;
    }

    /** Returns the elected child {@link LoadBalancerProvider}. */
    LoadBalancerProvider getEffectiveLbProvider() {
      return effectiveLbProvider;
    }

    @Override
    public boolean equals(Object o) {
      if (this == o) {
        return true;
      }
      if (o == null || getClass() != o.getClass()) {
        return false;
      }
      ChildLoadBalancingPolicy that = (ChildLoadBalancingPolicy) o;
      return Objects.equals(effectiveRawChildPolicy, that.effectiveRawChildPolicy)
          && Objects.equals(effectiveLbProvider, that.effectiveLbProvider)
          && Objects.equals(targetFieldName, that.targetFieldName);
    }

    @Override
    public int hashCode() {
      return Objects.hash(effectiveRawChildPolicy, effectiveLbProvider, targetFieldName);
    }

    @Override
    public String toString() {
      return MoreObjects.toStringHelper(this)
          .add("effectiveRawChildPolicy", effectiveRawChildPolicy)
          .add("effectiveLbProvider", effectiveLbProvider)
          .add("childPolicyConfigTargetFieldName", targetFieldName)
          .toString();
    }
  }

  /** Factory for {@link ChildPolicyWrapper}. Not thread-safe. */
  static final class RefCountedChildPolicyWrapperFactory {
    @VisibleForTesting
    final Map<String /* target */, RefCountedChildPolicyWrapper> childPolicyMap =
        new HashMap<>();

    private final ChildLoadBalancerHelperProvider childLbHelperProvider;
    private final ChildLoadBalancingPolicy childPolicy;
    private ResolvedAddressFactory childLbResolvedAddressFactory;

    public RefCountedChildPolicyWrapperFactory(
        ChildLoadBalancingPolicy childPolicy,
        ResolvedAddressFactory childLbResolvedAddressFactory,
        ChildLoadBalancerHelperProvider childLbHelperProvider) {
      this.childPolicy = checkNotNull(childPolicy, "childPolicy");
      this.childLbResolvedAddressFactory =
          checkNotNull(childLbResolvedAddressFactory, "childLbResolvedAddressFactory");
      this.childLbHelperProvider = checkNotNull(childLbHelperProvider, "childLbHelperProvider");
    }

    void init() {
      childLbHelperProvider.init();
    }

    Status acceptResolvedAddressFactory(ResolvedAddressFactory childLbResolvedAddressFactory) {
      this.childLbResolvedAddressFactory = childLbResolvedAddressFactory;
      Status status = Status.OK;
      for (RefCountedChildPolicyWrapper wrapper : childPolicyMap.values()) {
        Status newStatus =
            wrapper.childPolicyWrapper.acceptResolvedAddressFactory(childLbResolvedAddressFactory);
        if (!newStatus.isOk()) {
          status = newStatus;
        }
      }
      return status;
    }

    ChildPolicyWrapper createOrGet(String target) {
      // TODO(creamsoup) check if the target is valid or not
      RefCountedChildPolicyWrapper pooledChildPolicyWrapper = childPolicyMap.get(target);
      if (pooledChildPolicyWrapper == null) {
        ChildPolicyWrapper childPolicyWrapper = new ChildPolicyWrapper(
            target, childPolicy, childLbResolvedAddressFactory, childLbHelperProvider);
        pooledChildPolicyWrapper = RefCountedChildPolicyWrapper.of(childPolicyWrapper);
        childPolicyMap.put(target, pooledChildPolicyWrapper);
        return pooledChildPolicyWrapper.getObject();
      } else {
        ChildPolicyWrapper childPolicyWrapper = pooledChildPolicyWrapper.getObject();
        if (childPolicyWrapper.getPicker() != null) {
          childPolicyWrapper.refreshState();
        }
        return childPolicyWrapper;
      }
    }

    List<ChildPolicyWrapper> createOrGet(List<String> targets) {
      List<ChildPolicyWrapper> retVal = new ArrayList<>();
      for (String target : targets) {
        retVal.add(createOrGet(target));
      }
      return retVal;
    }

    void release(ChildPolicyWrapper childPolicyWrapper) {
      checkNotNull(childPolicyWrapper, "childPolicyWrapper");
      String target = childPolicyWrapper.getTarget();
      RefCountedChildPolicyWrapper existing = childPolicyMap.get(target);
      checkState(existing != null, "Cannot access already released object");
      existing.returnObject(childPolicyWrapper);
      if (existing.isReleased()) {
        childPolicyMap.remove(target);
      }
    }
  }

  /**
   * ChildPolicyWrapper is a wrapper class for child load balancing policy with associated helper /
   * utility classes to manage the child policy.
   */
  static final class ChildPolicyWrapper {

    private final String target;
    private final ChildPolicyReportingHelper helper;
    private final LoadBalancer lb;
    private final Object childLbConfig;
    private volatile SubchannelPicker picker;
    private ConnectivityState state;

    public ChildPolicyWrapper(
        String target,
        ChildLoadBalancingPolicy childPolicy,
        final ResolvedAddressFactory childLbResolvedAddressFactory,
        ChildLoadBalancerHelperProvider childLbHelperProvider) {
      this.target = target;
      this.helper = new ChildPolicyReportingHelper(childLbHelperProvider);
      LoadBalancerProvider lbProvider = childPolicy.getEffectiveLbProvider();
      final ConfigOrError lbConfig =
          lbProvider
              .parseLoadBalancingPolicyConfig(
                  childPolicy.getEffectiveChildPolicy(target));
      this.lb = lbProvider.newLoadBalancer(helper);
      this.childLbConfig = lbConfig.getConfig();
      helper.getChannelLogger().log(
          ChannelLogLevel.DEBUG, "RLS child lb created. config: {0}", childLbConfig);
      helper.getSynchronizationContext().execute(
          new Runnable() {
            @Override
            public void run() {
              if (!acceptResolvedAddressFactory(childLbResolvedAddressFactory).isOk()) {
                helper.refreshNameResolution();
              }
              lb.requestConnection();
            }
          });
    }

    Status acceptResolvedAddressFactory(ResolvedAddressFactory childLbResolvedAddressFactory) {
      helper.getSynchronizationContext().throwIfNotInThisSynchronizationContext();
      return lb.acceptResolvedAddresses(childLbResolvedAddressFactory.create(childLbConfig));
    }

    String getTarget() {
      return target;
    }

    SubchannelPicker getPicker() {
      return picker;
    }

    ChildPolicyReportingHelper getHelper() {
      return helper;
    }

    public ConnectivityState getState() {
      return state;
    }

    void refreshState() {
      helper.getSynchronizationContext().execute(
          new Runnable() {
            @Override
            public void run() {
              helper.updateBalancingState(state, picker);
            }
          }
      );
    }

    void shutdown() {
      helper.getSynchronizationContext().execute(
          new Runnable() {
            @Override
            public void run() {
              lb.shutdown();
            }
          }
      );
    }

    @Override
    public String toString() {
      return MoreObjects.toStringHelper(this)
          .add("target", target)
          .add("picker", picker)
          .add("state", state)
          .toString();
    }

    /**
     * A delegating {@link io.grpc.LoadBalancer.Helper} maintains status of {@link
     * ChildPolicyWrapper} when {@link Subchannel} status changed. This helper is used between child
     * policy and parent load-balancer where each picker in child policy is governed by a governing
     * picker (RlsPicker). The governing picker will be reported back to the parent load-balancer.
     */
    final class ChildPolicyReportingHelper extends ForwardingLoadBalancerHelper {

      private final ChildLoadBalancerHelper delegate;

      ChildPolicyReportingHelper(
          ChildLoadBalancerHelperProvider childHelperProvider) {
        checkNotNull(childHelperProvider, "childHelperProvider");
        this.delegate = childHelperProvider.forTarget(getTarget());
      }

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

      @Override
      public void updateBalancingState(ConnectivityState newState, SubchannelPicker newPicker) {
        picker = newPicker;
        state = newState;
        super.updateBalancingState(newState, newPicker);
      }
    }
  }

  private static final class RefCountedChildPolicyWrapper
      implements ObjectPool<ChildPolicyWrapper> {

    private long refCnt;
    @Nullable
    private ChildPolicyWrapper childPolicyWrapper;

    private RefCountedChildPolicyWrapper(ChildPolicyWrapper childPolicyWrapper) {
      this.childPolicyWrapper = checkNotNull(childPolicyWrapper, "childPolicyWrapper");
    }

    @Override
    public ChildPolicyWrapper getObject() {
      checkState(!isReleased(), "ChildPolicyWrapper is already released");
      refCnt++;
      return childPolicyWrapper;
    }

    @Override
    @Nullable
    public ChildPolicyWrapper returnObject(Object object) {
      checkState(
          !isReleased(),
          "cannot return already released ChildPolicyWrapper, this is possibly a bug.");
      checkState(
          childPolicyWrapper == object,
          "returned object doesn't match the pooled childPolicyWrapper");
      long newCnt = --refCnt;
      checkState(newCnt != -1, "Cannot return never pooled childPolicyWrapper");
      if (newCnt == 0) {
        childPolicyWrapper.shutdown();
        childPolicyWrapper = null;
      }
      return null;
    }

    boolean isReleased() {
      return childPolicyWrapper == null;
    }

    static RefCountedChildPolicyWrapper of(ChildPolicyWrapper childPolicyWrapper) {
      return new RefCountedChildPolicyWrapper(childPolicyWrapper);
    }

    @Override
    public String toString() {
      return MoreObjects.toStringHelper(this)
          .add("object", childPolicyWrapper)
          .add("refCnt", refCnt)
          .toString();
    }
  }

  /** Exception thrown when attempting to parse child policy encountered parsing issue. */
  static final class InvalidChildPolicyConfigException extends Exception {

    private static final long serialVersionUID = 0L;

    InvalidChildPolicyConfigException(String message) {
      super(message);
    }

    @Override
    public synchronized Throwable fillInStackTrace() {
      // no stack trace above this point
      return this;
    }
  }
}

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.rls;
18
19	import static com.google.common.base.Preconditions.checkArgument;
20	import static com.google.common.base.Preconditions.checkNotNull;
21	import static com.google.common.base.Preconditions.checkState;
22
23	import com.google.common.annotations.VisibleForTesting;
24	import com.google.common.base.MoreObjects;
25	import io.grpc.ChannelLogger.ChannelLogLevel;
26	import io.grpc.ConnectivityState;
27	import io.grpc.LoadBalancer;
28	import io.grpc.LoadBalancer.Helper;
29	import io.grpc.LoadBalancer.Subchannel;
30	import io.grpc.LoadBalancer.SubchannelPicker;
31	import io.grpc.LoadBalancerProvider;
32	import io.grpc.LoadBalancerRegistry;
33	import io.grpc.NameResolver.ConfigOrError;
34	import io.grpc.Status;
35	import io.grpc.internal.ObjectPool;
36	import io.grpc.rls.ChildLoadBalancerHelper.ChildLoadBalancerHelperProvider;
37	import io.grpc.rls.RlsProtoData.RouteLookupConfig;
38	import io.grpc.util.ForwardingLoadBalancerHelper;
39	import java.util.ArrayList;
40	import java.util.Collections;
41	import java.util.HashMap;
42	import java.util.List;
43	import java.util.Map;
44	import java.util.Objects;
45	import javax.annotation.Nullable;
46
47	/** Configuration for RLS load balancing policy. */
48	final class LbPolicyConfiguration {
49
50	private final RouteLookupConfig routeLookupConfig;
51	@Nullable
52	private final Map<String, ?> routeLookupChannelServiceConfig;
53	private final ChildLoadBalancingPolicy policy;
54
55	LbPolicyConfiguration(
56	RouteLookupConfig routeLookupConfig, @Nullable Map<String, ?> routeLookupChannelServiceConfig,
57	ChildLoadBalancingPolicy policy) {	1✔
58	this.routeLookupConfig = checkNotNull(routeLookupConfig, "routeLookupConfig");	1✔
59	this.routeLookupChannelServiceConfig = routeLookupChannelServiceConfig;	1✔
60	this.policy = checkNotNull(policy, "policy");	1✔
61	}	1✔
62
63	RouteLookupConfig getRouteLookupConfig() {
64	return routeLookupConfig;	1✔
65	}
66
67	@Nullable
68	Map<String, ?> getRouteLookupChannelServiceConfig() {
69	return routeLookupChannelServiceConfig;	1✔
70	}
71
72	ChildLoadBalancingPolicy getLoadBalancingPolicy() {
73	return policy;	1✔
74	}
75
76	@Override
77	public boolean equals(Object o) {
78	if (this == o) {	1✔
79	return true;	×
80	}
81	if (o == null \|\| getClass() != o.getClass()) {	1✔
82	return false;	1✔
83	}
84	LbPolicyConfiguration that = (LbPolicyConfiguration) o;	×
85	return Objects.equals(routeLookupConfig, that.routeLookupConfig)	×
86	&& Objects.equals(routeLookupChannelServiceConfig, that.routeLookupChannelServiceConfig)	×
87	&& Objects.equals(policy, that.policy);	×
88	}
89
90	@Override
91	public int hashCode() {
92	return Objects.hash(routeLookupConfig, routeLookupChannelServiceConfig, policy);	×
93	}
94
95	@Override
96	public String toString() {
97	return MoreObjects.toStringHelper(this)	×
98	.add("routeLookupConfig", routeLookupConfig)	×
99	.add("routeLookupChannelServiceConfig", routeLookupChannelServiceConfig)	×
100	.add("policy", policy)	×
101	.toString();	×
102	}
103
104	/** ChildLoadBalancingPolicy is an elected child policy to delegate requests. */
105	static final class ChildLoadBalancingPolicy {
106
107	private final Map<String, Object> effectiveRawChildPolicy;
108	private final LoadBalancerProvider effectiveLbProvider;
109	private final String targetFieldName;
110
111	@VisibleForTesting
112	ChildLoadBalancingPolicy(
113	String targetFieldName,
114	Map<String, Object> effectiveRawChildPolicy,
115	LoadBalancerProvider effectiveLbProvider) {	1✔
116	checkArgument(	1✔
117	targetFieldName != null && !targetFieldName.isEmpty(),	1✔
118	"targetFieldName cannot be empty or null");
119	this.targetFieldName = targetFieldName;	1✔
120	this.effectiveRawChildPolicy =	1✔
121	checkNotNull(effectiveRawChildPolicy, "effectiveRawChildPolicy");	1✔
122	this.effectiveLbProvider = checkNotNull(effectiveLbProvider, "effectiveLbProvider");	1✔
123	}	1✔
124
125	/** Creates ChildLoadBalancingPolicy. */
126	@SuppressWarnings("unchecked")
127	static ChildLoadBalancingPolicy create(
128	String childPolicyConfigTargetFieldName, List<Map<String, ?>> childPolicies)
129	throws InvalidChildPolicyConfigException {
130	Map<String, Object> effectiveChildPolicy = null;	1✔
131	LoadBalancerProvider effectiveLbProvider = null;	1✔
132	List<String> policyTried = new ArrayList<>();	1✔
133
134	LoadBalancerRegistry lbRegistry = LoadBalancerRegistry.getDefaultRegistry();	1✔
135	for (Map<String, ?> childPolicy : childPolicies) {	1✔
136	if (childPolicy.isEmpty()) {	1✔
137	continue;	×
138	}
139	if (childPolicy.size() != 1) {	1✔
140	throw	1✔
141	new InvalidChildPolicyConfigException(
142	"childPolicy should have exactly one loadbalancing policy");
143	}
144	String policyName = childPolicy.keySet().iterator().next();	1✔
145	LoadBalancerProvider provider = lbRegistry.getProvider(policyName);	1✔
146	if (provider != null) {	1✔
147	effectiveLbProvider = provider;	1✔
148	effectiveChildPolicy = Collections.unmodifiableMap(childPolicy);	1✔
149	break;	1✔
150	}
151	policyTried.add(policyName);	×
152	}	×
153	if (effectiveChildPolicy == null) {	1✔
154	throw	1✔
155	new InvalidChildPolicyConfigException(
156	String.format("no valid childPolicy found, policy tried: %s", policyTried));	1✔
157	}
158	return	1✔
159	new ChildLoadBalancingPolicy(
160	childPolicyConfigTargetFieldName,
161	(Map<String, Object>) effectiveChildPolicy.values().iterator().next(),	1✔
162	effectiveLbProvider);
163	}
164
165	/** Creates a child load balancer config for given target from elected raw child policy. */
166	Map<String, ?> getEffectiveChildPolicy(String target) {
167	Map<String, Object> childPolicy = new HashMap<>(effectiveRawChildPolicy);	1✔
168	childPolicy.put(targetFieldName, target);	1✔
169	return childPolicy;	1✔
170	}
171
172	/** Returns the elected child {@link LoadBalancerProvider}. */
173	LoadBalancerProvider getEffectiveLbProvider() {
174	return effectiveLbProvider;	1✔
175	}
176
177	@Override
178	public boolean equals(Object o) {
179	if (this == o) {	×
180	return true;	×
181	}
182	if (o == null \|\| getClass() != o.getClass()) {	×
183	return false;	×
184	}
185	ChildLoadBalancingPolicy that = (ChildLoadBalancingPolicy) o;	×
186	return Objects.equals(effectiveRawChildPolicy, that.effectiveRawChildPolicy)	×
187	&& Objects.equals(effectiveLbProvider, that.effectiveLbProvider)	×
188	&& Objects.equals(targetFieldName, that.targetFieldName);	×
189	}
190
191	@Override
192	public int hashCode() {
193	return Objects.hash(effectiveRawChildPolicy, effectiveLbProvider, targetFieldName);	×
194	}
195
196	@Override
197	public String toString() {
198	return MoreObjects.toStringHelper(this)	×
199	.add("effectiveRawChildPolicy", effectiveRawChildPolicy)	×
200	.add("effectiveLbProvider", effectiveLbProvider)	×
201	.add("childPolicyConfigTargetFieldName", targetFieldName)	×
202	.toString();	×
203	}
204	}
205
206	/** Factory for {@link ChildPolicyWrapper}. Not thread-safe. */
207	static final class RefCountedChildPolicyWrapperFactory {
208	@VisibleForTesting	1✔
209	final Map<String /* target */, RefCountedChildPolicyWrapper> childPolicyMap =
210	new HashMap<>();
211
212	private final ChildLoadBalancerHelperProvider childLbHelperProvider;
213	private final ChildLoadBalancingPolicy childPolicy;
214	private ResolvedAddressFactory childLbResolvedAddressFactory;
215
216	public RefCountedChildPolicyWrapperFactory(
217	ChildLoadBalancingPolicy childPolicy,
218	ResolvedAddressFactory childLbResolvedAddressFactory,
219	ChildLoadBalancerHelperProvider childLbHelperProvider) {	1✔
220	this.childPolicy = checkNotNull(childPolicy, "childPolicy");	1✔
221	this.childLbResolvedAddressFactory =	1✔
222	checkNotNull(childLbResolvedAddressFactory, "childLbResolvedAddressFactory");	1✔
223	this.childLbHelperProvider = checkNotNull(childLbHelperProvider, "childLbHelperProvider");	1✔
224	}	1✔
225
226	void init() {
227	childLbHelperProvider.init();	1✔
228	}	1✔
229
230	Status acceptResolvedAddressFactory(ResolvedAddressFactory childLbResolvedAddressFactory) {
231	this.childLbResolvedAddressFactory = childLbResolvedAddressFactory;	1✔
232	Status status = Status.OK;	1✔
233	for (RefCountedChildPolicyWrapper wrapper : childPolicyMap.values()) {	1✔
234	Status newStatus =	1✔
235	wrapper.childPolicyWrapper.acceptResolvedAddressFactory(childLbResolvedAddressFactory);	1✔
236	if (!newStatus.isOk()) {	1✔
237	status = newStatus;	×
238	}
239	}	1✔
240	return status;	1✔
241	}
242
243	ChildPolicyWrapper createOrGet(String target) {
244	// TODO(creamsoup) check if the target is valid or not
245	RefCountedChildPolicyWrapper pooledChildPolicyWrapper = childPolicyMap.get(target);	1✔
246	if (pooledChildPolicyWrapper == null) {	1✔
247	ChildPolicyWrapper childPolicyWrapper = new ChildPolicyWrapper(	1✔
248	target, childPolicy, childLbResolvedAddressFactory, childLbHelperProvider);
249	pooledChildPolicyWrapper = RefCountedChildPolicyWrapper.of(childPolicyWrapper);	1✔
250	childPolicyMap.put(target, pooledChildPolicyWrapper);	1✔
251	return pooledChildPolicyWrapper.getObject();	1✔
252	} else {
253	ChildPolicyWrapper childPolicyWrapper = pooledChildPolicyWrapper.getObject();	1✔
254	if (childPolicyWrapper.getPicker() != null) {	1✔
255	childPolicyWrapper.refreshState();	1✔
256	}
257	return childPolicyWrapper;	1✔
258	}
259	}
260
261	List<ChildPolicyWrapper> createOrGet(List<String> targets) {
262	List<ChildPolicyWrapper> retVal = new ArrayList<>();	1✔
263	for (String target : targets) {	1✔
264	retVal.add(createOrGet(target));	1✔
265	}	1✔
266	return retVal;	1✔
267	}
268
269	void release(ChildPolicyWrapper childPolicyWrapper) {
270	checkNotNull(childPolicyWrapper, "childPolicyWrapper");	1✔
271	String target = childPolicyWrapper.getTarget();	1✔
272	RefCountedChildPolicyWrapper existing = childPolicyMap.get(target);	1✔
273	checkState(existing != null, "Cannot access already released object");	1✔
274	existing.returnObject(childPolicyWrapper);	1✔
275	if (existing.isReleased()) {	1✔
276	childPolicyMap.remove(target);	1✔
277	}
278	}	1✔
279	}
280
281	/**
282	* ChildPolicyWrapper is a wrapper class for child load balancing policy with associated helper /
283	* utility classes to manage the child policy.
284	*/
285	static final class ChildPolicyWrapper {
286
287	private final String target;
288	private final ChildPolicyReportingHelper helper;
289	private final LoadBalancer lb;
290	private final Object childLbConfig;
291	private volatile SubchannelPicker picker;
292	private ConnectivityState state;
293
294	public ChildPolicyWrapper(
295	String target,
296	ChildLoadBalancingPolicy childPolicy,
297	final ResolvedAddressFactory childLbResolvedAddressFactory,
298	ChildLoadBalancerHelperProvider childLbHelperProvider) {	1✔
299	this.target = target;	1✔
300	this.helper = new ChildPolicyReportingHelper(childLbHelperProvider);	1✔
301	LoadBalancerProvider lbProvider = childPolicy.getEffectiveLbProvider();	1✔
302	final ConfigOrError lbConfig =	1✔
303	lbProvider
304	.parseLoadBalancingPolicyConfig(	1✔
305	childPolicy.getEffectiveChildPolicy(target));	1✔
306	this.lb = lbProvider.newLoadBalancer(helper);	1✔
307	this.childLbConfig = lbConfig.getConfig();	1✔
308	helper.getChannelLogger().log(	1✔
309	ChannelLogLevel.DEBUG, "RLS child lb created. config: {0}", childLbConfig);
310	helper.getSynchronizationContext().execute(	1✔
311	new Runnable() {	1✔
312	@Override
313	public void run() {
314	if (!acceptResolvedAddressFactory(childLbResolvedAddressFactory).isOk()) {	1✔
315	helper.refreshNameResolution();	1✔
316	}
317	lb.requestConnection();	1✔
318	}	1✔
319	});
320	}	1✔
321
322	Status acceptResolvedAddressFactory(ResolvedAddressFactory childLbResolvedAddressFactory) {
323	helper.getSynchronizationContext().throwIfNotInThisSynchronizationContext();	1✔
324	return lb.acceptResolvedAddresses(childLbResolvedAddressFactory.create(childLbConfig));	1✔
325	}
326
327	String getTarget() {
328	return target;	1✔
329	}
330
331	SubchannelPicker getPicker() {
332	return picker;	1✔
333	}
334
335	ChildPolicyReportingHelper getHelper() {
336	return helper;	1✔
337	}
338
339	public ConnectivityState getState() {
340	return state;	1✔
341	}
342
343	void refreshState() {
344	helper.getSynchronizationContext().execute(	1✔
345	new Runnable() {	1✔
346	@Override
347	public void run() {
348	helper.updateBalancingState(state, picker);	1✔
349	}	1✔
350	}
351	);
352	}	1✔
353
354	void shutdown() {
355	helper.getSynchronizationContext().execute(	1✔
356	new Runnable() {	1✔
357	@Override
358	public void run() {
359	lb.shutdown();	1✔
360	}	1✔
361	}
362	);
363	}	1✔
364
365	@Override
366	public String toString() {
367	return MoreObjects.toStringHelper(this)	×
368	.add("target", target)	×
369	.add("picker", picker)	×
370	.add("state", state)	×
371	.toString();	×
372	}
373
374	/**
375	* A delegating {@link io.grpc.LoadBalancer.Helper} maintains status of {@link
376	* ChildPolicyWrapper} when {@link Subchannel} status changed. This helper is used between child
377	* policy and parent load-balancer where each picker in child policy is governed by a governing
378	* picker (RlsPicker). The governing picker will be reported back to the parent load-balancer.
379	*/
380	final class ChildPolicyReportingHelper extends ForwardingLoadBalancerHelper {
381
382	private final ChildLoadBalancerHelper delegate;
383
384	ChildPolicyReportingHelper(
385	ChildLoadBalancerHelperProvider childHelperProvider) {	1✔
386	checkNotNull(childHelperProvider, "childHelperProvider");	1✔
387	this.delegate = childHelperProvider.forTarget(getTarget());	1✔
388	}	1✔
389
390	@Override
391	protected Helper delegate() {
392	return delegate;	1✔
393	}
394
395	@Override
396	public void updateBalancingState(ConnectivityState newState, SubchannelPicker newPicker) {
397	picker = newPicker;	1✔
398	state = newState;	1✔
399	super.updateBalancingState(newState, newPicker);	1✔
400	}	1✔
401	}
402	}
403
404	private static final class RefCountedChildPolicyWrapper
405	implements ObjectPool<ChildPolicyWrapper> {
406
407	private long refCnt;
408	@Nullable
409	private ChildPolicyWrapper childPolicyWrapper;
410
411	private RefCountedChildPolicyWrapper(ChildPolicyWrapper childPolicyWrapper) {	1✔
412	this.childPolicyWrapper = checkNotNull(childPolicyWrapper, "childPolicyWrapper");	1✔
413	}	1✔
414
415	@Override
416	public ChildPolicyWrapper getObject() {
417	checkState(!isReleased(), "ChildPolicyWrapper is already released");	1✔
418	refCnt++;	1✔
419	return childPolicyWrapper;	1✔
420	}
421
422	@Override
423	@Nullable
424	public ChildPolicyWrapper returnObject(Object object) {
425	checkState(	1✔
426	!isReleased(),	1✔
427	"cannot return already released ChildPolicyWrapper, this is possibly a bug.");
428	checkState(	1✔
429	childPolicyWrapper == object,
430	"returned object doesn't match the pooled childPolicyWrapper");
431	long newCnt = --refCnt;	1✔
432	checkState(newCnt != -1, "Cannot return never pooled childPolicyWrapper");	1✔
433	if (newCnt == 0) {	1✔
434	childPolicyWrapper.shutdown();	1✔
435	childPolicyWrapper = null;	1✔
436	}
437	return null;	1✔
438	}
439
440	boolean isReleased() {
441	return childPolicyWrapper == null;	1✔
442	}
443
444	static RefCountedChildPolicyWrapper of(ChildPolicyWrapper childPolicyWrapper) {
445	return new RefCountedChildPolicyWrapper(childPolicyWrapper);	1✔
446	}
447
448	@Override
449	public String toString() {
450	return MoreObjects.toStringHelper(this)	×
451	.add("object", childPolicyWrapper)	×
452	.add("refCnt", refCnt)	×
453	.toString();	×
454	}
455	}
456
457	/** Exception thrown when attempting to parse child policy encountered parsing issue. */
458	static final class InvalidChildPolicyConfigException extends Exception {
459
460	private static final long serialVersionUID = 0L;
461
462	InvalidChildPolicyConfigException(String message) {
463	super(message);	1✔
464	}	1✔
465
466	@Override
467	public synchronized Throwable fillInStackTrace() {
468	// no stack trace above this point
469	return this;	1✔
470	}
471	}
472	}

grpc / grpc-java / #20057

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