1 |
|
2 |
|
3 |
|
4 |
|
5 |
|
6 |
|
7 |
|
8 |
|
9 |
|
10 |
|
11 |
|
12 |
|
13 |
|
14 |
|
15 |
|
16 |
|
17 |
|
18 | import {
|
19 | ChannelControlHelper,
|
20 | LoadBalancer,
|
21 | LoadBalancingConfig,
|
22 | getFirstUsableConfig,
|
23 | } from './load-balancer';
|
24 | import { ServiceConfig, validateServiceConfig } from './service-config';
|
25 | import { ConnectivityState } from './connectivity-state';
|
26 | import { ConfigSelector, createResolver, Resolver } from './resolver';
|
27 | import { ServiceError } from './call';
|
28 | import { Picker, UnavailablePicker, QueuePicker } from './picker';
|
29 | import { BackoffOptions, BackoffTimeout } from './backoff-timeout';
|
30 | import { Status } from './constants';
|
31 | import { StatusObject } from './call-stream';
|
32 | import { Metadata } from './metadata';
|
33 | import * as logging from './logging';
|
34 | import { LogVerbosity } from './constants';
|
35 | import { SubchannelAddress } from './subchannel-address';
|
36 | import { GrpcUri, uriToString } from './uri-parser';
|
37 | import { ChildLoadBalancerHandler } from './load-balancer-child-handler';
|
38 | import { ChannelOptions } from './channel-options';
|
39 | import { PickFirstLoadBalancingConfig } from './load-balancer-pick-first';
|
40 |
|
41 | const TRACER_NAME = 'resolving_load_balancer';
|
42 |
|
43 | function trace(text: string): void {
|
44 | logging.trace(LogVerbosity.DEBUG, TRACER_NAME, text);
|
45 | }
|
46 |
|
47 | const DEFAULT_LOAD_BALANCER_NAME = 'pick_first';
|
48 |
|
49 | function getDefaultConfigSelector(
|
50 | serviceConfig: ServiceConfig | null
|
51 | ): ConfigSelector {
|
52 | return function defaultConfigSelector(
|
53 | methodName: string,
|
54 | metadata: Metadata
|
55 | ) {
|
56 | const splitName = methodName.split('/').filter((x) => x.length > 0);
|
57 | const service = splitName[0] ?? '';
|
58 | const method = splitName[1] ?? '';
|
59 | if (serviceConfig && serviceConfig.methodConfig) {
|
60 | for (const methodConfig of serviceConfig.methodConfig) {
|
61 | for (const name of methodConfig.name) {
|
62 | if (
|
63 | name.service === service &&
|
64 | (name.method === undefined || name.method === method)
|
65 | ) {
|
66 | return {
|
67 | methodConfig: methodConfig,
|
68 | pickInformation: {},
|
69 | status: Status.OK,
|
70 | dynamicFilterFactories: []
|
71 | };
|
72 | }
|
73 | }
|
74 | }
|
75 | }
|
76 | return {
|
77 | methodConfig: { name: [] },
|
78 | pickInformation: {},
|
79 | status: Status.OK,
|
80 | dynamicFilterFactories: []
|
81 | };
|
82 | };
|
83 | }
|
84 |
|
85 | export interface ResolutionCallback {
|
86 | (configSelector: ConfigSelector): void;
|
87 | }
|
88 |
|
89 | export interface ResolutionFailureCallback {
|
90 | (status: StatusObject): void;
|
91 | }
|
92 |
|
93 | export class ResolvingLoadBalancer implements LoadBalancer {
|
94 | |
95 |
|
96 |
|
97 | private innerResolver: Resolver;
|
98 |
|
99 | private childLoadBalancer: ChildLoadBalancerHandler;
|
100 | private latestChildState: ConnectivityState = ConnectivityState.IDLE;
|
101 | private latestChildPicker: Picker = new QueuePicker(this);
|
102 | |
103 |
|
104 |
|
105 | private currentState: ConnectivityState = ConnectivityState.IDLE;
|
106 | private readonly defaultServiceConfig: ServiceConfig;
|
107 | |
108 |
|
109 |
|
110 |
|
111 |
|
112 | private previousServiceConfig: ServiceConfig | null = null;
|
113 |
|
114 | |
115 |
|
116 |
|
117 | private readonly backoffTimeout: BackoffTimeout;
|
118 |
|
119 | |
120 |
|
121 |
|
122 |
|
123 | private continueResolving = false;
|
124 |
|
125 | |
126 |
|
127 |
|
128 |
|
129 |
|
130 |
|
131 |
|
132 |
|
133 |
|
134 |
|
135 |
|
136 |
|
137 | constructor(
|
138 | private readonly target: GrpcUri,
|
139 | private readonly channelControlHelper: ChannelControlHelper,
|
140 | private readonly channelOptions: ChannelOptions,
|
141 | private readonly onSuccessfulResolution: ResolutionCallback,
|
142 | private readonly onFailedResolution: ResolutionFailureCallback
|
143 | ) {
|
144 | if (channelOptions['grpc.service_config']) {
|
145 | this.defaultServiceConfig = validateServiceConfig(
|
146 | JSON.parse(channelOptions['grpc.service_config']!)
|
147 | );
|
148 | } else {
|
149 | this.defaultServiceConfig = {
|
150 | loadBalancingConfig: [],
|
151 | methodConfig: [],
|
152 | };
|
153 | }
|
154 | this.updateState(ConnectivityState.IDLE, new QueuePicker(this));
|
155 | this.childLoadBalancer = new ChildLoadBalancerHandler({
|
156 | createSubchannel: channelControlHelper.createSubchannel.bind(
|
157 | channelControlHelper
|
158 | ),
|
159 | requestReresolution: () => {
|
160 | |
161 |
|
162 |
|
163 |
|
164 | if (this.backoffTimeout.isRunning()) {
|
165 | this.continueResolving = true;
|
166 | } else {
|
167 | this.updateResolution();
|
168 | }
|
169 | },
|
170 | updateState: (newState: ConnectivityState, picker: Picker) => {
|
171 | this.latestChildState = newState;
|
172 | this.latestChildPicker = picker;
|
173 | this.updateState(newState, picker);
|
174 | },
|
175 | addChannelzChild: channelControlHelper.addChannelzChild.bind(
|
176 | channelControlHelper
|
177 | ),
|
178 | removeChannelzChild: channelControlHelper.removeChannelzChild.bind(
|
179 | channelControlHelper
|
180 | )
|
181 | });
|
182 | this.innerResolver = createResolver(
|
183 | target,
|
184 | {
|
185 | onSuccessfulResolution: (
|
186 | addressList: SubchannelAddress[],
|
187 | serviceConfig: ServiceConfig | null,
|
188 | serviceConfigError: ServiceError | null,
|
189 | configSelector: ConfigSelector | null,
|
190 | attributes: { [key: string]: unknown }
|
191 | ) => {
|
192 | let workingServiceConfig: ServiceConfig | null = null;
|
193 | |
194 |
|
195 |
|
196 |
|
197 | if (serviceConfig === null) {
|
198 |
|
199 | if (serviceConfigError === null) {
|
200 |
|
201 | this.previousServiceConfig = null;
|
202 | workingServiceConfig = this.defaultServiceConfig;
|
203 | } else {
|
204 |
|
205 | if (this.previousServiceConfig === null) {
|
206 |
|
207 | this.handleResolutionFailure(serviceConfigError);
|
208 | } else {
|
209 |
|
210 | workingServiceConfig = this.previousServiceConfig;
|
211 | }
|
212 | }
|
213 | } else {
|
214 |
|
215 | workingServiceConfig = serviceConfig;
|
216 | this.previousServiceConfig = serviceConfig;
|
217 | }
|
218 | const workingConfigList =
|
219 | workingServiceConfig?.loadBalancingConfig ?? [];
|
220 | const loadBalancingConfig = getFirstUsableConfig(
|
221 | workingConfigList,
|
222 | true
|
223 | );
|
224 | if (loadBalancingConfig === null) {
|
225 |
|
226 | this.handleResolutionFailure({
|
227 | code: Status.UNAVAILABLE,
|
228 | details:
|
229 | 'All load balancer options in service config are not compatible',
|
230 | metadata: new Metadata(),
|
231 | });
|
232 | return;
|
233 | }
|
234 | this.childLoadBalancer.updateAddressList(
|
235 | addressList,
|
236 | loadBalancingConfig,
|
237 | attributes
|
238 | );
|
239 | const finalServiceConfig =
|
240 | workingServiceConfig ?? this.defaultServiceConfig;
|
241 | this.onSuccessfulResolution(
|
242 | configSelector ?? getDefaultConfigSelector(finalServiceConfig)
|
243 | );
|
244 | },
|
245 | onError: (error: StatusObject) => {
|
246 | this.handleResolutionFailure(error);
|
247 | },
|
248 | },
|
249 | channelOptions
|
250 | );
|
251 | const backoffOptions: BackoffOptions = {
|
252 | initialDelay: channelOptions['grpc.initial_reconnect_backoff_ms'],
|
253 | maxDelay: channelOptions['grpc.max_reconnect_backoff_ms'],
|
254 | };
|
255 | this.backoffTimeout = new BackoffTimeout(() => {
|
256 | if (this.continueResolving) {
|
257 | this.updateResolution();
|
258 | this.continueResolving = false;
|
259 | } else {
|
260 | this.updateState(this.latestChildState, this.latestChildPicker);
|
261 | }
|
262 | }, backoffOptions);
|
263 | this.backoffTimeout.unref();
|
264 | }
|
265 |
|
266 | private updateResolution() {
|
267 | this.innerResolver.updateResolution();
|
268 | if (this.currentState === ConnectivityState.IDLE) {
|
269 | this.updateState(ConnectivityState.CONNECTING, new QueuePicker(this));
|
270 | }
|
271 | this.backoffTimeout.runOnce();
|
272 | }
|
273 |
|
274 | private updateState(connectivityState: ConnectivityState, picker: Picker) {
|
275 | trace(
|
276 | uriToString(this.target) +
|
277 | ' ' +
|
278 | ConnectivityState[this.currentState] +
|
279 | ' -> ' +
|
280 | ConnectivityState[connectivityState]
|
281 | );
|
282 |
|
283 | if (connectivityState === ConnectivityState.IDLE) {
|
284 | picker = new QueuePicker(this);
|
285 | }
|
286 | this.currentState = connectivityState;
|
287 | this.channelControlHelper.updateState(connectivityState, picker);
|
288 | }
|
289 |
|
290 | private handleResolutionFailure(error: StatusObject) {
|
291 | if (this.latestChildState === ConnectivityState.IDLE) {
|
292 | this.updateState(
|
293 | ConnectivityState.TRANSIENT_FAILURE,
|
294 | new UnavailablePicker(error)
|
295 | );
|
296 | this.onFailedResolution(error);
|
297 | }
|
298 | }
|
299 |
|
300 | exitIdle() {
|
301 | if (this.currentState === ConnectivityState.IDLE || this.currentState === ConnectivityState.TRANSIENT_FAILURE) {
|
302 | if (this.backoffTimeout.isRunning()) {
|
303 | this.continueResolving = true;
|
304 | } else {
|
305 | this.updateResolution();
|
306 | }
|
307 | }
|
308 | this.childLoadBalancer.exitIdle();
|
309 | }
|
310 |
|
311 | updateAddressList(
|
312 | addressList: SubchannelAddress[],
|
313 | lbConfig: LoadBalancingConfig | null
|
314 | ): never {
|
315 | throw new Error('updateAddressList not supported on ResolvingLoadBalancer');
|
316 | }
|
317 |
|
318 | resetBackoff() {
|
319 | this.backoffTimeout.reset();
|
320 | this.childLoadBalancer.resetBackoff();
|
321 | }
|
322 |
|
323 | destroy() {
|
324 | this.childLoadBalancer.destroy();
|
325 | this.innerResolver.destroy();
|
326 | this.updateState(ConnectivityState.SHUTDOWN, new UnavailablePicker());
|
327 | }
|
328 |
|
329 | getTypeName() {
|
330 | return 'resolving_load_balancer';
|
331 | }
|
332 | }
|