#19900

Committed 09 Jul 2025 02:53PM UTC coverage: 88.528% (-0.001%) from 88.529%

Build # #19900

Build Type

push

github

Committed by

ejona86

Commit Message

Fix RLS regressions from XdsDepMan conversion

297ab05ef converted CDS to XdsDependencyManager. This caused three
regressions:

 * CdsLB2 as a RLS child would always fail with "Unable to find
   non-dynamic root cluster" because is_dynamic=true was missing in
   its service config
 * XdsNameResolver only propagated resolution updates when the clusters
   changed, so a CdsUpdate change would be ignored. This caused a hang
   for RLS even with is_dynamic=true. For non-RLS the lack config update
   broke the circuit breaking psm interop test. This would have been
   more severe if ClusterResolverLb had been converted to
   XdsDependenceManager, as it would have ignored EDS updates
 * RLS did not propagate resolution updates, so CdsLB2 even with
   is_dynamic=true the CdsUpdate for the new cluster would never arrive,
   causing a hang

b/428120265
b/427912384

Run Details

34649 of 39139 relevant lines covered (88.53%)

0.89 hits per line

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

77.3

/../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 ChildLbStatusListener childLbStatusListener;
    private final ChildLoadBalancingPolicy childPolicy;
    private ResolvedAddressFactory childLbResolvedAddressFactory;

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

    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,
            childLbStatusListener);
        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,
        ChildLbStatusListener childLbStatusListener) {
      this.target = target;
      this.helper =
          new ChildPolicyReportingHelper(childLbHelperProvider, childLbStatusListener);
      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;
      private final ChildLbStatusListener listener;

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

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

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

  /** Listener for child lb status change events. */
  interface ChildLbStatusListener {

    /** Notifies when child lb status changes. */
    void onStatusChanged(ConnectivityState newState);
  }

  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 ChildLbStatusListener childLbStatusListener;
214	private final ChildLoadBalancingPolicy childPolicy;
215	private ResolvedAddressFactory childLbResolvedAddressFactory;
216
217	public RefCountedChildPolicyWrapperFactory(
218	ChildLoadBalancingPolicy childPolicy,
219	ResolvedAddressFactory childLbResolvedAddressFactory,
220	ChildLoadBalancerHelperProvider childLbHelperProvider,
221	ChildLbStatusListener childLbStatusListener) {	1✔
222	this.childPolicy = checkNotNull(childPolicy, "childPolicy");	1✔
223	this.childLbResolvedAddressFactory =	1✔
224	checkNotNull(childLbResolvedAddressFactory, "childLbResolvedAddressFactory");	1✔
225	this.childLbHelperProvider = checkNotNull(childLbHelperProvider, "childLbHelperProvider");	1✔
226	this.childLbStatusListener = checkNotNull(childLbStatusListener, "childLbStatusListener");	1✔
227	}	1✔
228
229	void init() {
230	childLbHelperProvider.init();	1✔
231	}	1✔
232
233	Status acceptResolvedAddressFactory(ResolvedAddressFactory childLbResolvedAddressFactory) {
234	this.childLbResolvedAddressFactory = childLbResolvedAddressFactory;	1✔
235	Status status = Status.OK;	1✔
236	for (RefCountedChildPolicyWrapper wrapper : childPolicyMap.values()) {	1✔
237	Status newStatus =	×
238	wrapper.childPolicyWrapper.acceptResolvedAddressFactory(childLbResolvedAddressFactory);	×
239	if (!newStatus.isOk()) {	×
240	status = newStatus;	×
241	}
242	}	×
243	return status;	1✔
244	}
245
246	ChildPolicyWrapper createOrGet(String target) {
247	// TODO(creamsoup) check if the target is valid or not
248	RefCountedChildPolicyWrapper pooledChildPolicyWrapper = childPolicyMap.get(target);	1✔
249	if (pooledChildPolicyWrapper == null) {	1✔
250	ChildPolicyWrapper childPolicyWrapper = new ChildPolicyWrapper(	1✔
251	target, childPolicy, childLbResolvedAddressFactory, childLbHelperProvider,
252	childLbStatusListener);
253	pooledChildPolicyWrapper = RefCountedChildPolicyWrapper.of(childPolicyWrapper);	1✔
254	childPolicyMap.put(target, pooledChildPolicyWrapper);	1✔
255	return pooledChildPolicyWrapper.getObject();	1✔
256	} else {
257	ChildPolicyWrapper childPolicyWrapper = pooledChildPolicyWrapper.getObject();	1✔
258	if (childPolicyWrapper.getPicker() != null) {	1✔
259	childPolicyWrapper.refreshState();	1✔
260	}
261	return childPolicyWrapper;	1✔
262	}
263	}
264
265	List<ChildPolicyWrapper> createOrGet(List<String> targets) {
266	List<ChildPolicyWrapper> retVal = new ArrayList<>();	1✔
267	for (String target : targets) {	1✔
268	retVal.add(createOrGet(target));	1✔
269	}	1✔
270	return retVal;	1✔
271	}
272
273	void release(ChildPolicyWrapper childPolicyWrapper) {
274	checkNotNull(childPolicyWrapper, "childPolicyWrapper");	1✔
275	String target = childPolicyWrapper.getTarget();	1✔
276	RefCountedChildPolicyWrapper existing = childPolicyMap.get(target);	1✔
277	checkState(existing != null, "Cannot access already released object");	1✔
278	existing.returnObject(childPolicyWrapper);	1✔
279	if (existing.isReleased()) {	1✔
280	childPolicyMap.remove(target);	1✔
281	}
282	}	1✔
283	}
284
285	/**
286	* ChildPolicyWrapper is a wrapper class for child load balancing policy with associated helper /
287	* utility classes to manage the child policy.
288	*/
289	static final class ChildPolicyWrapper {
290
291	private final String target;
292	private final ChildPolicyReportingHelper helper;
293	private final LoadBalancer lb;
294	private final Object childLbConfig;
295	private volatile SubchannelPicker picker;
296	private ConnectivityState state;
297
298	public ChildPolicyWrapper(
299	String target,
300	ChildLoadBalancingPolicy childPolicy,
301	final ResolvedAddressFactory childLbResolvedAddressFactory,
302	ChildLoadBalancerHelperProvider childLbHelperProvider,
303	ChildLbStatusListener childLbStatusListener) {	1✔
304	this.target = target;	1✔
305	this.helper =	1✔
306	new ChildPolicyReportingHelper(childLbHelperProvider, childLbStatusListener);
307	LoadBalancerProvider lbProvider = childPolicy.getEffectiveLbProvider();	1✔
308	final ConfigOrError lbConfig =	1✔
309	lbProvider
310	.parseLoadBalancingPolicyConfig(	1✔
311	childPolicy.getEffectiveChildPolicy(target));	1✔
312	this.lb = lbProvider.newLoadBalancer(helper);	1✔
313	this.childLbConfig = lbConfig.getConfig();	1✔
314	helper.getChannelLogger().log(	1✔
315	ChannelLogLevel.DEBUG, "RLS child lb created. config: {0}", childLbConfig);
316	helper.getSynchronizationContext().execute(	1✔
317	new Runnable() {	1✔
318	@Override
319	public void run() {
320	if (!acceptResolvedAddressFactory(childLbResolvedAddressFactory).isOk()) {	1✔
321	helper.refreshNameResolution();	1✔
322	}
323	lb.requestConnection();	1✔
324	}	1✔
325	});
326	}	1✔
327
328	Status acceptResolvedAddressFactory(ResolvedAddressFactory childLbResolvedAddressFactory) {
329	helper.getSynchronizationContext().throwIfNotInThisSynchronizationContext();	1✔
330	return lb.acceptResolvedAddresses(childLbResolvedAddressFactory.create(childLbConfig));	1✔
331	}
332
333	String getTarget() {
334	return target;	1✔
335	}
336
337	SubchannelPicker getPicker() {
338	return picker;	1✔
339	}
340
341	ChildPolicyReportingHelper getHelper() {
342	return helper;	1✔
343	}
344
345	public ConnectivityState getState() {
346	return state;	1✔
347	}
348
349	void refreshState() {
350	helper.getSynchronizationContext().execute(	1✔
351	new Runnable() {	1✔
352	@Override
353	public void run() {
354	helper.updateBalancingState(state, picker);	1✔
355	}	1✔
356	}
357	);
358	}	1✔
359
360	void shutdown() {
361	helper.getSynchronizationContext().execute(	1✔
362	new Runnable() {	1✔
363	@Override
364	public void run() {
365	lb.shutdown();	1✔
366	}	1✔
367	}
368	);
369	}	1✔
370
371	@Override
372	public String toString() {
373	return MoreObjects.toStringHelper(this)	×
374	.add("target", target)	×
375	.add("picker", picker)	×
376	.add("state", state)	×
377	.toString();	×
378	}
379
380	/**
381	* A delegating {@link io.grpc.LoadBalancer.Helper} maintains status of {@link
382	* ChildPolicyWrapper} when {@link Subchannel} status changed. This helper is used between child
383	* policy and parent load-balancer where each picker in child policy is governed by a governing
384	* picker (RlsPicker). The governing picker will be reported back to the parent load-balancer.
385	*/
386	final class ChildPolicyReportingHelper extends ForwardingLoadBalancerHelper {
387
388	private final ChildLoadBalancerHelper delegate;
389	private final ChildLbStatusListener listener;
390
391	ChildPolicyReportingHelper(
392	ChildLoadBalancerHelperProvider childHelperProvider,
393	ChildLbStatusListener listener) {	1✔
394	checkNotNull(childHelperProvider, "childHelperProvider");	1✔
395	this.delegate = childHelperProvider.forTarget(getTarget());	1✔
396	this.listener = checkNotNull(listener, "listener");	1✔
397	}	1✔
398
399	@Override
400	protected Helper delegate() {
401	return delegate;	1✔
402	}
403
404	@Override
405	public void updateBalancingState(ConnectivityState newState, SubchannelPicker newPicker) {
406	picker = newPicker;	1✔
407	state = newState;	1✔
408	super.updateBalancingState(newState, newPicker);	1✔
409	listener.onStatusChanged(newState);	1✔
410	}	1✔
411	}
412	}
413
414	/** Listener for child lb status change events. */
415	interface ChildLbStatusListener {
416
417	/** Notifies when child lb status changes. */
418	void onStatusChanged(ConnectivityState newState);
419	}
420
421	private static final class RefCountedChildPolicyWrapper
422	implements ObjectPool<ChildPolicyWrapper> {
423
424	private long refCnt;
425	@Nullable
426	private ChildPolicyWrapper childPolicyWrapper;
427
428	private RefCountedChildPolicyWrapper(ChildPolicyWrapper childPolicyWrapper) {	1✔
429	this.childPolicyWrapper = checkNotNull(childPolicyWrapper, "childPolicyWrapper");	1✔
430	}	1✔
431
432	@Override
433	public ChildPolicyWrapper getObject() {
434	checkState(!isReleased(), "ChildPolicyWrapper is already released");	1✔
435	refCnt++;	1✔
436	return childPolicyWrapper;	1✔
437	}
438
439	@Override
440	@Nullable
441	public ChildPolicyWrapper returnObject(Object object) {
442	checkState(	1✔
443	!isReleased(),	1✔
444	"cannot return already released ChildPolicyWrapper, this is possibly a bug.");
445	checkState(	1✔
446	childPolicyWrapper == object,
447	"returned object doesn't match the pooled childPolicyWrapper");
448	long newCnt = --refCnt;	1✔
449	checkState(newCnt != -1, "Cannot return never pooled childPolicyWrapper");	1✔
450	if (newCnt == 0) {	1✔
451	childPolicyWrapper.shutdown();	1✔
452	childPolicyWrapper = null;	1✔
453	}
454	return null;	1✔
455	}
456
457	boolean isReleased() {
458	return childPolicyWrapper == null;	1✔
459	}
460
461	static RefCountedChildPolicyWrapper of(ChildPolicyWrapper childPolicyWrapper) {
462	return new RefCountedChildPolicyWrapper(childPolicyWrapper);	1✔
463	}
464
465	@Override
466	public String toString() {
467	return MoreObjects.toStringHelper(this)	×
468	.add("object", childPolicyWrapper)	×
469	.add("refCnt", refCnt)	×
470	.toString();	×
471	}
472	}
473
474	/** Exception thrown when attempting to parse child policy encountered parsing issue. */
475	static final class InvalidChildPolicyConfigException extends Exception {
476
477	private static final long serialVersionUID = 0L;
478
479	InvalidChildPolicyConfigException(String message) {
480	super(message);	1✔
481	}	1✔
482
483	@Override
484	public synchronized Throwable fillInStackTrace() {
485	// no stack trace above this point
486	return this;	1✔
487	}
488	}
489	}

grpc / grpc-java / #19900

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