• Home
  • Features
  • Pricing
  • Docs
  • Announcements
  • Sign In

grpc / grpc-java / #19682

11 Feb 2025 01:14AM CUT coverage: 88.605% (-0.004%) from 88.609%
#19682

push

github

web-flow
xds: improve code readability of server FilterChain parsing

- Improve code flow and variable names
- Reduce nesting
- Add comments between logical blocks
- Add comments explaining some xDS/gRPC nuances

34182 of 38578 relevant lines covered (88.6%)

0.89 hits per line

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

93.72
/../xds/src/main/java/io/grpc/xds/client/XdsClientImpl.java
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.client;
18

19
import static com.google.common.base.Preconditions.checkArgument;
20
import static com.google.common.base.Preconditions.checkNotNull;
21
import static io.grpc.xds.client.Bootstrapper.XDSTP_SCHEME;
22
import static io.grpc.xds.client.XdsResourceType.ParsedResource;
23
import static io.grpc.xds.client.XdsResourceType.ValidatedResourceUpdate;
24

25
import com.google.common.annotations.VisibleForTesting;
26
import com.google.common.base.Joiner;
27
import com.google.common.base.Stopwatch;
28
import com.google.common.base.Supplier;
29
import com.google.common.collect.ImmutableList;
30
import com.google.common.collect.ImmutableMap;
31
import com.google.common.util.concurrent.ListenableFuture;
32
import com.google.common.util.concurrent.SettableFuture;
33
import com.google.protobuf.Any;
34
import io.grpc.Internal;
35
import io.grpc.InternalLogId;
36
import io.grpc.Status;
37
import io.grpc.SynchronizationContext;
38
import io.grpc.SynchronizationContext.ScheduledHandle;
39
import io.grpc.internal.BackoffPolicy;
40
import io.grpc.internal.TimeProvider;
41
import io.grpc.xds.client.Bootstrapper.AuthorityInfo;
42
import io.grpc.xds.client.Bootstrapper.ServerInfo;
43
import io.grpc.xds.client.XdsClient.ResourceStore;
44
import io.grpc.xds.client.XdsLogger.XdsLogLevel;
45
import java.io.IOException;
46
import java.net.URI;
47
import java.util.ArrayList;
48
import java.util.Collection;
49
import java.util.Collections;
50
import java.util.HashMap;
51
import java.util.HashSet;
52
import java.util.List;
53
import java.util.Map;
54
import java.util.Objects;
55
import java.util.Set;
56
import java.util.concurrent.Executor;
57
import java.util.concurrent.Future;
58
import java.util.concurrent.ScheduledExecutorService;
59
import java.util.concurrent.TimeUnit;
60
import java.util.stream.Collectors;
61
import javax.annotation.Nullable;
62

63
/**
64
 * XdsClient implementation.
65
 */
66
@Internal
67
public final class XdsClientImpl extends XdsClient implements ResourceStore {
68

69
  // Longest time to wait, since the subscription to some resource, for concluding its absence.
70
  @VisibleForTesting
71
  public static final int INITIAL_RESOURCE_FETCH_TIMEOUT_SEC = 15;
72

73
  private final SynchronizationContext syncContext = new SynchronizationContext(
1✔
74
      new Thread.UncaughtExceptionHandler() {
1✔
75
        @Override
76
        public void uncaughtException(Thread t, Throwable e) {
77
          logger.log(
×
78
              XdsLogLevel.ERROR,
79
              "Uncaught exception in XdsClient SynchronizationContext. Panic!",
80
              e);
81
          // TODO: better error handling.
82
          throw new AssertionError(e);
×
83
        }
84
      });
85

86
  private final Map<ServerInfo, LoadStatsManager2> loadStatsManagerMap = new HashMap<>();
1✔
87
  final Map<ServerInfo, LoadReportClient> serverLrsClientMap = new HashMap<>();
1✔
88
  /** Map of authority to its activated control plane client (affected by xds fallback).
89
   * The last entry in the list for each value is the "active" CPC for the matching key */
90
  private final Map<String, List<ControlPlaneClient>> activatedCpClients = new HashMap<>();
1✔
91
  private final Map<ServerInfo, ControlPlaneClient> serverCpClientMap = new HashMap<>();
1✔
92

93
  /** Maps resource type to the corresponding map of subscribers (keyed by resource name). */
94
  private final Map<XdsResourceType<? extends ResourceUpdate>,
1✔
95
      Map<String, ResourceSubscriber<? extends ResourceUpdate>>>
96
      resourceSubscribers = new HashMap<>();
97
  /** Maps typeUrl to the corresponding XdsResourceType. */
98
  private final Map<String, XdsResourceType<?>> subscribedResourceTypeUrls = new HashMap<>();
1✔
99

100
  private final XdsTransportFactory xdsTransportFactory;
101
  private final Bootstrapper.BootstrapInfo bootstrapInfo;
102
  private final ScheduledExecutorService timeService;
103
  private final BackoffPolicy.Provider backoffPolicyProvider;
104
  private final Supplier<Stopwatch> stopwatchSupplier;
105
  private final TimeProvider timeProvider;
106
  private final Object securityConfig;
107
  private final InternalLogId logId;
108
  private final XdsLogger logger;
109
  private volatile boolean isShutdown;
110
  private final MessagePrettyPrinter messagePrinter;
111
  private final XdsClientMetricReporter metricReporter;
112

113
  public XdsClientImpl(
114
      XdsTransportFactory xdsTransportFactory,
115
      Bootstrapper.BootstrapInfo bootstrapInfo,
116
      ScheduledExecutorService timeService,
117
      BackoffPolicy.Provider backoffPolicyProvider,
118
      Supplier<Stopwatch> stopwatchSupplier,
119
      TimeProvider timeProvider,
120
      MessagePrettyPrinter messagePrinter,
121
      Object securityConfig,
122
      XdsClientMetricReporter metricReporter) {
1✔
123
    this.xdsTransportFactory = xdsTransportFactory;
1✔
124
    this.bootstrapInfo = bootstrapInfo;
1✔
125
    this.timeService = timeService;
1✔
126
    this.backoffPolicyProvider = backoffPolicyProvider;
1✔
127
    this.stopwatchSupplier = stopwatchSupplier;
1✔
128
    this.timeProvider = timeProvider;
1✔
129
    this.messagePrinter = messagePrinter;
1✔
130
    this.securityConfig = securityConfig;
1✔
131
    this.metricReporter = metricReporter;
1✔
132
    logId = InternalLogId.allocate("xds-client", null);
1✔
133
    logger = XdsLogger.withLogId(logId);
1✔
134
    logger.log(XdsLogLevel.INFO, "Created");
1✔
135
  }
1✔
136

137
  @Override
138
  public void shutdown() {
139
    syncContext.execute(
1✔
140
        new Runnable() {
1✔
141
          @Override
142
          public void run() {
143
            if (isShutdown) {
1✔
144
              return;
1✔
145
            }
146
            isShutdown = true;
1✔
147
            for (ControlPlaneClient xdsChannel : serverCpClientMap.values()) {
1✔
148
              xdsChannel.shutdown();
1✔
149
            }
1✔
150
            for (final LoadReportClient lrsClient : serverLrsClientMap.values()) {
1✔
151
              lrsClient.stopLoadReporting();
1✔
152
            }
1✔
153
            cleanUpResourceTimers(null);
1✔
154
            activatedCpClients.clear();
1✔
155
          }
1✔
156
        });
157
  }
1✔
158

159
  @Override
160
  public boolean isShutDown() {
161
    return isShutdown;
1✔
162
  }
163

164
  @Override
165
  public Map<String, XdsResourceType<?>> getSubscribedResourceTypesWithTypeUrl() {
166
    return Collections.unmodifiableMap(subscribedResourceTypeUrls);
1✔
167
  }
168

169
  private ControlPlaneClient getActiveCpc(String authority) {
170
    List<ControlPlaneClient> controlPlaneClients = activatedCpClients.get(authority);
1✔
171
    if (controlPlaneClients == null || controlPlaneClients.isEmpty()) {
1✔
172
      return null;
1✔
173
    }
174

175
    return controlPlaneClients.get(controlPlaneClients.size() - 1);
1✔
176
  }
177

178
  @Nullable
179
  @Override
180
  public Collection<String> getSubscribedResources(
181
      ServerInfo serverInfo, XdsResourceType<? extends ResourceUpdate> type) {
182
    ControlPlaneClient targetCpc = serverCpClientMap.get(serverInfo);
1✔
183
    if (targetCpc == null) {
1✔
184
      return null;
×
185
    }
186

187
    // This should include all of the authorities that targetCpc or a fallback from it is serving
188
    List<String> authorities = activatedCpClients.entrySet().stream()
1✔
189
        .filter(entry -> entry.getValue().contains(targetCpc))
1✔
190
        .map(Map.Entry::getKey)
1✔
191
        .collect(Collectors.toList());
1✔
192

193
    Map<String, ResourceSubscriber<? extends ResourceUpdate>> resources =
1✔
194
        resourceSubscribers.getOrDefault(type, Collections.emptyMap());
1✔
195

196
    Collection<String> retVal = resources.entrySet().stream()
1✔
197
        .filter(entry -> authorities.contains(entry.getValue().authority))
1✔
198
        .map(Map.Entry::getKey)
1✔
199
        .collect(Collectors.toList());
1✔
200

201
    return retVal.isEmpty() ? null : retVal;
1✔
202
  }
203

204
  @Override
205
  public void startMissingResourceTimers(Collection<String> resourceNames,
206
                                         XdsResourceType<?> resourceType) {
207
    Map<String, ResourceSubscriber<? extends ResourceUpdate>> subscriberMap =
1✔
208
        resourceSubscribers.get(resourceType);
1✔
209

210
    for (String resourceName : resourceNames) {
1✔
211
      ResourceSubscriber<?> subscriber = subscriberMap.get(resourceName);
1✔
212
      if (subscriber.respTimer == null && !subscriber.hasResult()) {
1✔
213
        subscriber.restartTimer();
1✔
214
      }
215
    }
1✔
216
  }
1✔
217

218
  // As XdsClient APIs becomes resource agnostic, subscribed resource types are dynamic.
219
  // ResourceTypes that do not have subscribers does not show up in the snapshot keys.
220
  @Override
221
  public ListenableFuture<Map<XdsResourceType<?>, Map<String, ResourceMetadata>>>
222
      getSubscribedResourcesMetadataSnapshot() {
223
    final SettableFuture<Map<XdsResourceType<?>, Map<String, ResourceMetadata>>> future =
224
        SettableFuture.create();
1✔
225
    syncContext.execute(new Runnable() {
1✔
226
      @Override
227
      public void run() {
228
        // A map from a "resource type" to a map ("resource name": "resource metadata")
229
        ImmutableMap.Builder<XdsResourceType<?>, Map<String, ResourceMetadata>> metadataSnapshot =
230
            ImmutableMap.builder();
1✔
231
        for (XdsResourceType<?> resourceType : resourceSubscribers.keySet()) {
1✔
232
          ImmutableMap.Builder<String, ResourceMetadata> metadataMap = ImmutableMap.builder();
1✔
233
          for (Map.Entry<String, ResourceSubscriber<? extends ResourceUpdate>> resourceEntry
234
              : resourceSubscribers.get(resourceType).entrySet()) {
1✔
235
            metadataMap.put(resourceEntry.getKey(), resourceEntry.getValue().metadata);
1✔
236
          }
1✔
237
          metadataSnapshot.put(resourceType, metadataMap.buildOrThrow());
1✔
238
        }
1✔
239
        future.set(metadataSnapshot.buildOrThrow());
1✔
240
      }
1✔
241
    });
242
    return future;
1✔
243
  }
244

245
  @Override
246
  public Object getSecurityConfig() {
247
    return securityConfig;
×
248
  }
249

250
  @Override
251
  public <T extends ResourceUpdate> void watchXdsResource(XdsResourceType<T> type,
252
                                                          String resourceName,
253
                                                          ResourceWatcher<T> watcher,
254
                                                          Executor watcherExecutor) {
255
    syncContext.execute(new Runnable() {
1✔
256
      @Override
257
      @SuppressWarnings("unchecked")
258
      public void run() {
259
        if (!resourceSubscribers.containsKey(type)) {
1✔
260
          resourceSubscribers.put(type, new HashMap<>());
1✔
261
          subscribedResourceTypeUrls.put(type.typeUrl(), type);
1✔
262
        }
263
        ResourceSubscriber<T> subscriber =
1✔
264
            (ResourceSubscriber<T>) resourceSubscribers.get(type).get(resourceName);
1✔
265

266
        if (subscriber == null) {
1✔
267
          logger.log(XdsLogLevel.INFO, "Subscribe {0} resource {1}", type, resourceName);
1✔
268
          subscriber = new ResourceSubscriber<>(type, resourceName);
1✔
269
          resourceSubscribers.get(type).put(resourceName, subscriber);
1✔
270

271
          if (subscriber.errorDescription == null) {
1✔
272
            CpcWithFallbackState cpcToUse = manageControlPlaneClient(subscriber);
1✔
273
            if (cpcToUse.cpc != null) {
1✔
274
              cpcToUse.cpc.adjustResourceSubscription(type);
1✔
275
            }
276
          }
277
        }
278

279
        subscriber.addWatcher(watcher, watcherExecutor);
1✔
280
      }
1✔
281
    });
282
  }
1✔
283

284
  /**
285
   * Gets a ControlPlaneClient for the subscriber's authority, creating one if necessary.
286
   * If there already was an active CPC for this authority, and it is different from the one
287
   * identified, then do fallback to the identified one (cpcToUse).
288
   *
289
   * @return identified CPC or {@code null} (if there are no valid ServerInfos associated with the
290
   *     subscriber's authority or CPC's for all are in backoff), and whether did a fallback.
291
   */
292
  @VisibleForTesting
293
  private <T extends ResourceUpdate> CpcWithFallbackState manageControlPlaneClient(
294
      ResourceSubscriber<T> subscriber) {
295

296
    ControlPlaneClient cpcToUse;
297
    boolean didFallback = false;
1✔
298
    try {
299
      cpcToUse = getOrCreateControlPlaneClient(subscriber.authority);
1✔
300
    } catch (IllegalArgumentException e) {
×
301
      if (subscriber.errorDescription == null) {
×
302
        subscriber.errorDescription = "Bad configuration:  " + e.getMessage();
×
303
      }
304

305
      subscriber.onError(
×
306
          Status.INVALID_ARGUMENT.withDescription(subscriber.errorDescription), null);
×
307
      return new CpcWithFallbackState(null, false);
×
308
    } catch (IOException e) {
1✔
309
      logger.log(XdsLogLevel.DEBUG,
1✔
310
          "Could not create a control plane client for authority {0}: {1}",
311
          subscriber.authority, e.getMessage());
1✔
312
      return new CpcWithFallbackState(null, false);
1✔
313
    }
1✔
314

315
    ControlPlaneClient activeCpClient = getActiveCpc(subscriber.authority);
1✔
316
    if (cpcToUse != activeCpClient) {
1✔
317
      addCpcToAuthority(subscriber.authority, cpcToUse); // makes it active
1✔
318
      if (activeCpClient != null) {
1✔
319
        didFallback = cpcToUse != null && !cpcToUse.isInError();
1✔
320
        if (didFallback) {
1✔
321
          logger.log(XdsLogLevel.INFO, "Falling back to XDS server {0}",
1✔
322
              cpcToUse.getServerInfo().target());
1✔
323
        } else {
324
          logger.log(XdsLogLevel.WARNING, "No working fallback XDS Servers found from {0}",
×
325
              activeCpClient.getServerInfo().target());
×
326
        }
327
      }
328
    }
329

330
    return new CpcWithFallbackState(cpcToUse, didFallback);
1✔
331
  }
332

333
  private void addCpcToAuthority(String authority, ControlPlaneClient cpcToUse) {
334
    List<ControlPlaneClient> controlPlaneClients =
1✔
335
        activatedCpClients.computeIfAbsent(authority, k -> new ArrayList<>());
1✔
336

337
    if (controlPlaneClients.contains(cpcToUse)) {
1✔
338
      return;
×
339
    }
340

341
    // if there are any missing CPCs between the last one and cpcToUse, add them + add cpcToUse
342
    ImmutableList<ServerInfo> serverInfos = getServerInfos(authority);
1✔
343
    for (int i = controlPlaneClients.size(); i < serverInfos.size(); i++) {
1✔
344
      ServerInfo serverInfo = serverInfos.get(i);
1✔
345
      ControlPlaneClient cpc = serverCpClientMap.get(serverInfo);
1✔
346
      controlPlaneClients.add(cpc);
1✔
347
      logger.log(XdsLogLevel.DEBUG, "Adding control plane client {0} to authority {1}",
1✔
348
          cpc, authority);
349
      cpcToUse.sendDiscoveryRequests();
1✔
350
      if (cpc == cpcToUse) {
1✔
351
        break;
1✔
352
      }
353
    }
354
  }
1✔
355

356
  @Override
357
  public <T extends ResourceUpdate> void cancelXdsResourceWatch(XdsResourceType<T> type,
358
                                                                String resourceName,
359
                                                                ResourceWatcher<T> watcher) {
360
    syncContext.execute(new Runnable() {
1✔
361
      @Override
362
      @SuppressWarnings("unchecked")
363
      public void run() {
364
        ResourceSubscriber<T> subscriber =
1✔
365
            (ResourceSubscriber<T>) resourceSubscribers.get(type).get(resourceName);
1✔
366
        if (subscriber == null) {
1✔
367
          logger.log(XdsLogLevel.WARNING, "double cancel of resource watch for {0}:{1}",
×
368
              type.typeName(), resourceName);
×
369
          return;
×
370
        }
371
        subscriber.removeWatcher(watcher);
1✔
372
        if (!subscriber.isWatched()) {
1✔
373
          subscriber.cancelResourceWatch();
1✔
374
          resourceSubscribers.get(type).remove(resourceName);
1✔
375

376
          List<ControlPlaneClient> controlPlaneClients =
1✔
377
              activatedCpClients.get(subscriber.authority);
1✔
378
          if (controlPlaneClients != null) {
1✔
379
            controlPlaneClients.forEach((cpc) -> {
1✔
380
              cpc.adjustResourceSubscription(type);
1✔
381
            });
1✔
382
          }
383

384
          if (resourceSubscribers.get(type).isEmpty()) {
1✔
385
            resourceSubscribers.remove(type);
1✔
386
            subscribedResourceTypeUrls.remove(type.typeUrl());
1✔
387
          }
388
        }
389
      }
1✔
390
    });
391
  }
1✔
392

393
  @Override
394
  public LoadStatsManager2.ClusterDropStats addClusterDropStats(
395
      final ServerInfo serverInfo, String clusterName,
396
      @Nullable String edsServiceName) {
397
    LoadStatsManager2 loadStatsManager = loadStatsManagerMap.get(serverInfo);
1✔
398
    LoadStatsManager2.ClusterDropStats dropCounter =
1✔
399
        loadStatsManager.getClusterDropStats(clusterName, edsServiceName);
1✔
400
    syncContext.execute(new Runnable() {
1✔
401
      @Override
402
      public void run() {
403
        serverLrsClientMap.get(serverInfo).startLoadReporting();
1✔
404
      }
1✔
405
    });
406
    return dropCounter;
1✔
407
  }
408

409
  @Override
410
  public LoadStatsManager2.ClusterLocalityStats addClusterLocalityStats(
411
      final ServerInfo serverInfo, String clusterName, @Nullable String edsServiceName,
412
      Locality locality) {
413
    LoadStatsManager2 loadStatsManager = loadStatsManagerMap.get(serverInfo);
1✔
414
    LoadStatsManager2.ClusterLocalityStats loadCounter =
1✔
415
        loadStatsManager.getClusterLocalityStats(clusterName, edsServiceName, locality);
1✔
416
    syncContext.execute(new Runnable() {
1✔
417
      @Override
418
      public void run() {
419
        serverLrsClientMap.get(serverInfo).startLoadReporting();
1✔
420
      }
1✔
421
    });
422
    return loadCounter;
1✔
423
  }
424

425

426
  @Override
427
  public Bootstrapper.BootstrapInfo getBootstrapInfo() {
428
    return bootstrapInfo;
1✔
429
  }
430

431
  @Override
432
  public String toString() {
433
    return logId.toString();
×
434
  }
435

436
  private Set<String> getResourceKeys(XdsResourceType<?> xdsResourceType) {
437
    if (!resourceSubscribers.containsKey(xdsResourceType)) {
1✔
438
      return null;
×
439
    }
440

441
    return resourceSubscribers.get(xdsResourceType).keySet();
1✔
442
  }
443

444
  // cpcForThisStream is null when doing shutdown
445
  private void cleanUpResourceTimers(ControlPlaneClient cpcForThisStream) {
446
    Collection<String> authoritiesForCpc = getActiveAuthorities(cpcForThisStream);
1✔
447
    String target = cpcForThisStream == null ? "null" : cpcForThisStream.getServerInfo().target();
1✔
448
    logger.log(XdsLogLevel.DEBUG, "Cleaning up resource timers for CPC {0}, authorities {1}",
1✔
449
        target, authoritiesForCpc);
450

451
    for (Map<String, ResourceSubscriber<?>> subscriberMap : resourceSubscribers.values()) {
1✔
452
      for (ResourceSubscriber<?> subscriber : subscriberMap.values()) {
1✔
453
        if (cpcForThisStream == null || authoritiesForCpc.contains(subscriber.authority)) {
1✔
454
          subscriber.stopTimer();
1✔
455
        }
456
      }
1✔
457
    }
1✔
458
  }
1✔
459

460
  private ControlPlaneClient getOrCreateControlPlaneClient(String authority) throws IOException {
461
    // Optimize for the common case of a working ads stream already exists for the authority
462
    ControlPlaneClient activeCpc = getActiveCpc(authority);
1✔
463
    if (activeCpc != null && !activeCpc.isInError()) {
1✔
464
      return activeCpc;
1✔
465
    }
466

467
    ImmutableList<ServerInfo> serverInfos = getServerInfos(authority);
1✔
468
    if (serverInfos == null) {
1✔
469
      throw new IllegalArgumentException("No xds servers found for authority " + authority);
×
470
    }
471

472
    for (ServerInfo serverInfo : serverInfos) {
1✔
473
      ControlPlaneClient cpc = getOrCreateControlPlaneClient(serverInfo);
1✔
474
      if (cpc.isInError()) {
1✔
475
        continue;
1✔
476
      }
477
      return cpc;
1✔
478
    }
479

480
    // Everything existed and is in backoff so throw
481
    throw new IOException("All xds transports for authority " + authority + " are in backoff");
1✔
482
  }
483

484
  private ControlPlaneClient getOrCreateControlPlaneClient(ServerInfo serverInfo) {
485
    syncContext.throwIfNotInThisSynchronizationContext();
1✔
486
    if (serverCpClientMap.containsKey(serverInfo)) {
1✔
487
      return serverCpClientMap.get(serverInfo);
1✔
488
    }
489

490
    logger.log(XdsLogLevel.DEBUG, "Creating control plane client for {0}", serverInfo.target());
1✔
491
    XdsTransportFactory.XdsTransport xdsTransport;
492
    try {
493
      xdsTransport = xdsTransportFactory.create(serverInfo);
1✔
494
    } catch (Exception e) {
1✔
495
      String msg = String.format("Failed to create xds transport for %s: %s",
1✔
496
          serverInfo.target(), e.getMessage());
1✔
497
      logger.log(XdsLogLevel.WARNING, msg);
1✔
498
      xdsTransport =
1✔
499
          new ControlPlaneClient.FailingXdsTransport(Status.UNAVAILABLE.withDescription(msg));
1✔
500
    }
1✔
501

502
    ControlPlaneClient controlPlaneClient = new ControlPlaneClient(
1✔
503
        xdsTransport,
504
        serverInfo,
505
        bootstrapInfo.node(),
1✔
506
        new ResponseHandler(serverInfo),
507
        this,
508
        timeService,
509
        syncContext,
510
        backoffPolicyProvider,
511
        stopwatchSupplier,
512
        messagePrinter
513
    );
514

515
    serverCpClientMap.put(serverInfo, controlPlaneClient);
1✔
516

517
    LoadStatsManager2 loadStatsManager = new LoadStatsManager2(stopwatchSupplier);
1✔
518
    loadStatsManagerMap.put(serverInfo, loadStatsManager);
1✔
519
    LoadReportClient lrsClient = new LoadReportClient(
1✔
520
        loadStatsManager, xdsTransport, bootstrapInfo.node(),
1✔
521
        syncContext, timeService, backoffPolicyProvider, stopwatchSupplier);
522
    serverLrsClientMap.put(serverInfo, lrsClient);
1✔
523

524
    return controlPlaneClient;
1✔
525
  }
526

527
  @VisibleForTesting
528
  @Override
529
  public Map<ServerInfo, LoadReportClient> getServerLrsClientMap() {
530
    return ImmutableMap.copyOf(serverLrsClientMap);
1✔
531
  }
532

533
  private String getAuthority(String resource) {
534
    String authority;
535
    if (resource.startsWith(XDSTP_SCHEME)) {
1✔
536
      URI uri = URI.create(resource);
1✔
537
      authority = uri.getAuthority();
1✔
538
      if (authority == null) {
1✔
539
        authority = "";
1✔
540
      }
541
    } else {
1✔
542
      authority = null;
1✔
543
    }
544

545
    return authority;
1✔
546
  }
547

548
  @Nullable
549
  private ImmutableList<ServerInfo> getServerInfos(String authority) {
550
    if (authority != null) {
1✔
551
      AuthorityInfo authorityInfo = bootstrapInfo.authorities().get(authority);
1✔
552
      if (authorityInfo == null || authorityInfo.xdsServers().isEmpty()) {
1✔
553
        return null;
1✔
554
      }
555
      return authorityInfo.xdsServers();
1✔
556
    } else {
557
      return bootstrapInfo.servers();
1✔
558
    }
559
  }
560

561
  @SuppressWarnings("unchecked")
562
  private <T extends ResourceUpdate> void handleResourceUpdate(
563
      XdsResourceType.Args args, List<Any> resources, XdsResourceType<T> xdsResourceType,
564
      boolean isFirstResponse, ProcessingTracker processingTracker) {
565
    ControlPlaneClient controlPlaneClient = serverCpClientMap.get(args.serverInfo);
1✔
566

567
    if (isFirstResponse) {
1✔
568
      shutdownLowerPriorityCpcs(controlPlaneClient);
1✔
569
    }
570

571
    ValidatedResourceUpdate<T> result = xdsResourceType.parse(args, resources);
1✔
572
    logger.log(XdsLogger.XdsLogLevel.INFO,
1✔
573
        "Received {0} Response version {1} nonce {2}. Parsed resources: {3}",
574
        xdsResourceType.typeName(), args.versionInfo, args.nonce, result.unpackedResources);
1✔
575
    Map<String, ParsedResource<T>> parsedResources = result.parsedResources;
1✔
576
    Set<String> invalidResources = result.invalidResources;
1✔
577
    metricReporter.reportResourceUpdates(Long.valueOf(parsedResources.size()),
1✔
578
        Long.valueOf(invalidResources.size()),
1✔
579
        args.getServerInfo().target(), xdsResourceType.typeUrl());
1✔
580

581
    List<String> errors = result.errors;
1✔
582
    String errorDetail = null;
1✔
583
    if (errors.isEmpty()) {
1✔
584
      checkArgument(invalidResources.isEmpty(), "found invalid resources but missing errors");
1✔
585
      controlPlaneClient.ackResponse(xdsResourceType, args.versionInfo, args.nonce);
1✔
586
    } else {
587
      errorDetail = Joiner.on('\n').join(errors);
1✔
588
      logger.log(XdsLogLevel.WARNING,
1✔
589
          "Failed processing {0} Response version {1} nonce {2}. Errors:\n{3}",
590
          xdsResourceType.typeName(), args.versionInfo, args.nonce, errorDetail);
1✔
591
      controlPlaneClient.nackResponse(xdsResourceType, args.nonce, errorDetail);
1✔
592
    }
593

594
    long updateTime = timeProvider.currentTimeNanos();
1✔
595
    Map<String, ResourceSubscriber<? extends ResourceUpdate>> subscribedResources =
1✔
596
        resourceSubscribers.getOrDefault(xdsResourceType, Collections.emptyMap());
1✔
597
    for (Map.Entry<String, ResourceSubscriber<?>> entry : subscribedResources.entrySet()) {
1✔
598
      String resourceName = entry.getKey();
1✔
599
      ResourceSubscriber<T> subscriber = (ResourceSubscriber<T>) entry.getValue();
1✔
600
      if (parsedResources.containsKey(resourceName)) {
1✔
601
        // Happy path: the resource updated successfully. Notify the watchers of the update.
602
        subscriber.onData(parsedResources.get(resourceName), args.versionInfo, updateTime,
1✔
603
            processingTracker);
604
        continue;
1✔
605
      }
606

607
      if (invalidResources.contains(resourceName)) {
1✔
608
        // The resource update is invalid. Capture the error without notifying the watchers.
609
        subscriber.onRejected(args.versionInfo, updateTime, errorDetail);
1✔
610
      }
611

612
      // Nothing else to do for incremental ADS resources.
613
      if (!xdsResourceType.isFullStateOfTheWorld()) {
1✔
614
        continue;
1✔
615
      }
616

617
      // Handle State of the World ADS: invalid resources.
618
      if (invalidResources.contains(resourceName)) {
1✔
619
        // The resource is missing. Reuse the cached resource if possible.
620
        if (subscriber.data == null) {
1✔
621
          // No cached data. Notify the watchers of an invalid update.
622
          subscriber.onError(Status.UNAVAILABLE.withDescription(errorDetail), processingTracker);
1✔
623
        }
624
        continue;
625
      }
626

627
      // For State of the World services, notify watchers when their watched resource is missing
628
      // from the ADS update. Note that we can only do this if the resource update is coming from
629
      // the same xDS server that the ResourceSubscriber is subscribed to.
630
      if (getActiveCpc(subscriber.authority) == controlPlaneClient) {
1✔
631
        subscriber.onAbsent(processingTracker, args.serverInfo);
1✔
632
      }
633
    }
1✔
634
  }
1✔
635

636
  @Override
637
  public Future<Void> reportServerConnections(ServerConnectionCallback callback) {
638
    SettableFuture<Void> future = SettableFuture.create();
1✔
639
    syncContext.execute(() -> {
1✔
640
      serverCpClientMap.forEach((serverInfo, controlPlaneClient) ->
1✔
641
          callback.reportServerConnectionGauge(
1✔
642
              !controlPlaneClient.isInError(), serverInfo.target()));
1✔
643
      future.set(null);
1✔
644
    });
1✔
645
    return future;
1✔
646
  }
647

648
  private void shutdownLowerPriorityCpcs(ControlPlaneClient activatedCpc) {
649
    // For each authority, remove any control plane clients, with lower priority than the activated
650
    // one, from activatedCpClients storing them all in cpcsToShutdown.
651
    Set<ControlPlaneClient> cpcsToShutdown = new HashSet<>();
1✔
652
    for ( List<ControlPlaneClient> cpcsForAuth : activatedCpClients.values()) {
1✔
653
      if (cpcsForAuth == null) {
1✔
654
        continue;
×
655
      }
656
      int index = cpcsForAuth.indexOf(activatedCpc);
1✔
657
      if (index > -1) {
1✔
658
        cpcsToShutdown.addAll(cpcsForAuth.subList(index + 1, cpcsForAuth.size()));
1✔
659
        cpcsForAuth.subList(index + 1, cpcsForAuth.size()).clear(); // remove lower priority cpcs
1✔
660
      }
661
    }
1✔
662

663
    // Shutdown any lower priority control plane clients identified above that aren't still being
664
    // used by another authority.  If they are still being used let the XDS server know that we
665
    // no longer are interested in subscriptions for authorities we are no longer responsible for.
666
    for (ControlPlaneClient cpc : cpcsToShutdown) {
1✔
667
      if (activatedCpClients.values().stream().noneMatch(list -> list.contains(cpc))) {
1✔
668
        cpc.shutdown();
1✔
669
        serverCpClientMap.remove(cpc.getServerInfo());
1✔
670
      } else {
671
        cpc.sendDiscoveryRequests();
×
672
      }
673
    }
1✔
674
  }
1✔
675

676

677
  /** Tracks a single subscribed resource. */
678
  private final class ResourceSubscriber<T extends ResourceUpdate> {
679
    @Nullable
680
    private final String authority;
681
    private final XdsResourceType<T> type;
682
    private final String resource;
683
    private final Map<ResourceWatcher<T>, Executor> watchers = new HashMap<>();
1✔
684
    @Nullable
685
    private T data;
686
    private boolean absent;
687
    // Tracks whether the deletion has been ignored per bootstrap server feature.
688
    // See https://github.com/grpc/proposal/blob/master/A53-xds-ignore-resource-deletion.md
689
    private boolean resourceDeletionIgnored;
690
    @Nullable
691
    private ScheduledHandle respTimer;
692
    @Nullable
693
    private ResourceMetadata metadata;
694
    @Nullable
695
    private String errorDescription;
696

697
    ResourceSubscriber(XdsResourceType<T> type, String resource) {
1✔
698
      syncContext.throwIfNotInThisSynchronizationContext();
1✔
699
      this.type = type;
1✔
700
      this.resource = resource;
1✔
701
      this.authority = getAuthority(resource);
1✔
702
      if (getServerInfos(authority) == null) {
1✔
703
        this.errorDescription = "Wrong configuration: xds server does not exist for resource "
1✔
704
            + resource;
705
        return;
1✔
706
      }
707

708
      // Initialize metadata in UNKNOWN state to cover the case when resource subscriber,
709
      // is created but not yet requested because the client is in backoff.
710
      this.metadata = ResourceMetadata.newResourceMetadataUnknown();
1✔
711
    }
1✔
712

713
    @Override
714
    public String toString() {
715
      return "ResourceSubscriber{"
×
716
          + "resource='" + resource + '\''
717
          + ", authority='" + authority + '\''
718
          + ", type=" + type
719
          + ", watchers=" + watchers.size()
×
720
          + ", data=" + data
721
          + ", absent=" + absent
722
          + ", resourceDeletionIgnored=" + resourceDeletionIgnored
723
          + ", errorDescription='" + errorDescription + '\''
724
          + '}';
725
    }
726

727
    void addWatcher(ResourceWatcher<T> watcher, Executor watcherExecutor) {
728
      checkArgument(!watchers.containsKey(watcher), "watcher %s already registered", watcher);
1✔
729
      watchers.put(watcher, watcherExecutor);
1✔
730
      T savedData = data;
1✔
731
      boolean savedAbsent = absent;
1✔
732
      watcherExecutor.execute(() -> {
1✔
733
        if (errorDescription != null) {
1✔
734
          watcher.onError(Status.INVALID_ARGUMENT.withDescription(errorDescription));
1✔
735
          return;
1✔
736
        }
737
        if (savedData != null) {
1✔
738
          notifyWatcher(watcher, savedData);
1✔
739
        } else if (savedAbsent) {
1✔
740
          watcher.onResourceDoesNotExist(resource);
1✔
741
        }
742
      });
1✔
743
    }
1✔
744

745
    void removeWatcher(ResourceWatcher<T> watcher) {
746
      checkArgument(watchers.containsKey(watcher), "watcher %s not registered", watcher);
1✔
747
      watchers.remove(watcher);
1✔
748
    }
1✔
749

750
    void restartTimer() {
751
      if (data != null || absent) {  // resource already resolved
1✔
752
        return;
×
753
      }
754
      ControlPlaneClient activeCpc = getActiveCpc(authority);
1✔
755
      if (activeCpc == null || !activeCpc.isReady()) {
1✔
756
        // When client becomes ready, it triggers a restartTimer for all relevant subscribers.
757
        return;
1✔
758
      }
759

760
      class ResourceNotFound implements Runnable {
1✔
761
        @Override
762
        public void run() {
763
          logger.log(XdsLogLevel.INFO, "{0} resource {1} initial fetch timeout",
1✔
764
              type, resource);
1✔
765
          respTimer = null;
1✔
766
          onAbsent(null, activeCpc.getServerInfo());
1✔
767
        }
1✔
768

769
        @Override
770
        public String toString() {
771
          return type + this.getClass().getSimpleName();
1✔
772
        }
773
      }
774

775
      // Initial fetch scheduled or rescheduled, transition metadata state to REQUESTED.
776
      metadata = ResourceMetadata.newResourceMetadataRequested();
1✔
777

778
      if (respTimer != null) {
1✔
779
        respTimer.cancel();
×
780
      }
781
      respTimer = syncContext.schedule(
1✔
782
          new ResourceNotFound(), INITIAL_RESOURCE_FETCH_TIMEOUT_SEC, TimeUnit.SECONDS,
783
          timeService);
1✔
784
    }
1✔
785

786
    void stopTimer() {
787
      if (respTimer != null && respTimer.isPending()) {
1✔
788
        respTimer.cancel();
1✔
789
        respTimer = null;
1✔
790
      }
791
    }
1✔
792

793
    void cancelResourceWatch() {
794
      if (isWatched()) {
1✔
795
        throw new IllegalStateException("Can't cancel resource watch with active watchers present");
×
796
      }
797
      stopTimer();
1✔
798
      String message = "Unsubscribing {0} resource {1} from server {2}";
1✔
799
      XdsLogLevel logLevel = XdsLogLevel.INFO;
1✔
800
      if (resourceDeletionIgnored) {
1✔
801
        message += " for which we previously ignored a deletion";
×
802
        logLevel = XdsLogLevel.FORCE_INFO;
×
803
      }
804
      logger.log(logLevel, message, type, resource, getTarget());
1✔
805
    }
1✔
806

807
    boolean isWatched() {
808
      return !watchers.isEmpty();
1✔
809
    }
810

811
    boolean hasResult() {
812
      return data != null || absent;
1✔
813
    }
814

815
    void onData(ParsedResource<T> parsedResource, String version, long updateTime,
816
                ProcessingTracker processingTracker) {
817
      if (respTimer != null && respTimer.isPending()) {
1✔
818
        respTimer.cancel();
1✔
819
        respTimer = null;
1✔
820
      }
821
      ResourceUpdate oldData = this.data;
1✔
822
      this.data = parsedResource.getResourceUpdate();
1✔
823
      this.metadata = ResourceMetadata
1✔
824
          .newResourceMetadataAcked(parsedResource.getRawResource(), version, updateTime);
1✔
825
      absent = false;
1✔
826
      if (resourceDeletionIgnored) {
1✔
827
        logger.log(XdsLogLevel.FORCE_INFO, "xds server {0}: server returned new version "
1✔
828
                + "of resource for which we previously ignored a deletion: type {1} name {2}",
829
            getTarget(), type, resource);
1✔
830
        resourceDeletionIgnored = false;
1✔
831
      }
832
      if (!Objects.equals(oldData, data)) {
1✔
833
        for (ResourceWatcher<T> watcher : watchers.keySet()) {
1✔
834
          processingTracker.startTask();
1✔
835
          watchers.get(watcher).execute(() -> {
1✔
836
            try {
837
              notifyWatcher(watcher, data);
1✔
838
            } finally {
839
              processingTracker.onComplete();
1✔
840
            }
841
          });
1✔
842
        }
1✔
843
      }
844
    }
1✔
845

846
    private String getTarget() {
847
      ControlPlaneClient activeCpc = getActiveCpc(authority);
1✔
848
      return (activeCpc != null)
1✔
849
             ? activeCpc.getServerInfo().target()
1✔
850
             : "unknown";
1✔
851
    }
852

853
    void onAbsent(@Nullable ProcessingTracker processingTracker, ServerInfo serverInfo) {
854
      if (respTimer != null && respTimer.isPending()) {  // too early to conclude absence
1✔
855
        return;
1✔
856
      }
857

858
      // Ignore deletion of State of the World resources when this feature is on,
859
      // and the resource is reusable.
860
      boolean ignoreResourceDeletionEnabled = serverInfo.ignoreResourceDeletion();
1✔
861
      if (ignoreResourceDeletionEnabled && type.isFullStateOfTheWorld() && data != null) {
1✔
862
        if (!resourceDeletionIgnored) {
1✔
863
          logger.log(XdsLogLevel.FORCE_WARNING,
1✔
864
              "xds server {0}: ignoring deletion for resource type {1} name {2}}",
865
              serverInfo.target(), type, resource);
1✔
866
          resourceDeletionIgnored = true;
1✔
867
        }
868
        return;
1✔
869
      }
870

871
      logger.log(XdsLogLevel.INFO, "Conclude {0} resource {1} not exist", type, resource);
1✔
872
      if (!absent) {
1✔
873
        data = null;
1✔
874
        absent = true;
1✔
875
        metadata = ResourceMetadata.newResourceMetadataDoesNotExist();
1✔
876
        for (ResourceWatcher<T> watcher : watchers.keySet()) {
1✔
877
          if (processingTracker != null) {
1✔
878
            processingTracker.startTask();
1✔
879
          }
880
          watchers.get(watcher).execute(() -> {
1✔
881
            try {
882
              watcher.onResourceDoesNotExist(resource);
1✔
883
            } finally {
884
              if (processingTracker != null) {
1✔
885
                processingTracker.onComplete();
1✔
886
              }
887
            }
888
          });
1✔
889
        }
1✔
890
      }
891
    }
1✔
892

893
    void onError(Status error, @Nullable ProcessingTracker tracker) {
894
      if (respTimer != null && respTimer.isPending()) {
1✔
895
        respTimer.cancel();
1✔
896
        respTimer = null;
1✔
897
      }
898

899
      // Include node ID in xds failures to allow cross-referencing with control plane logs
900
      // when debugging.
901
      String description = error.getDescription() == null ? "" : error.getDescription() + " ";
1✔
902
      Status errorAugmented = Status.fromCode(error.getCode())
1✔
903
          .withDescription(description + "nodeID: " + bootstrapInfo.node().getId())
1✔
904
          .withCause(error.getCause());
1✔
905

906
      for (ResourceWatcher<T> watcher : watchers.keySet()) {
1✔
907
        if (tracker != null) {
1✔
908
          tracker.startTask();
1✔
909
        }
910
        watchers.get(watcher).execute(() -> {
1✔
911
          try {
912
            watcher.onError(errorAugmented);
1✔
913
          } finally {
914
            if (tracker != null) {
1✔
915
              tracker.onComplete();
1✔
916
            }
917
          }
918
        });
1✔
919
      }
1✔
920
    }
1✔
921

922
    void onRejected(String rejectedVersion, long rejectedTime, String rejectedDetails) {
923
      metadata = ResourceMetadata
1✔
924
          .newResourceMetadataNacked(metadata, rejectedVersion, rejectedTime, rejectedDetails,
1✔
925
              data != null);
926
    }
1✔
927

928
    private void notifyWatcher(ResourceWatcher<T> watcher, T update) {
929
      watcher.onChanged(update);
1✔
930
    }
1✔
931
  }
932

933
  private class ResponseHandler implements XdsResponseHandler {
934
    final ServerInfo serverInfo;
935

936
    ResponseHandler(ServerInfo serverInfo) {
1✔
937
      this.serverInfo = serverInfo;
1✔
938
    }
1✔
939

940
    @Override
941
    public void handleResourceResponse(
942
        XdsResourceType<?> xdsResourceType, ServerInfo serverInfo, String versionInfo,
943
        List<Any> resources, String nonce, boolean isFirstResponse,
944
        ProcessingTracker processingTracker) {
945
      checkNotNull(xdsResourceType, "xdsResourceType");
1✔
946
      syncContext.throwIfNotInThisSynchronizationContext();
1✔
947
      Set<String> toParseResourceNames =
948
          xdsResourceType.shouldRetrieveResourceKeysForArgs()
1✔
949
          ? getResourceKeys(xdsResourceType)
1✔
950
          : null;
1✔
951
      XdsResourceType.Args args = new XdsResourceType.Args(serverInfo, versionInfo, nonce,
1✔
952
          bootstrapInfo, securityConfig, toParseResourceNames);
1✔
953
      handleResourceUpdate(args, resources, xdsResourceType, isFirstResponse, processingTracker);
1✔
954
    }
1✔
955

956
    @Override
957
    public void handleStreamClosed(Status status, boolean shouldTryFallback) {
958
      syncContext.throwIfNotInThisSynchronizationContext();
1✔
959

960
      ControlPlaneClient cpcClosed = serverCpClientMap.get(serverInfo);
1✔
961
      if (cpcClosed == null) {
1✔
962
        logger.log(XdsLogLevel.DEBUG,
×
963
            "Couldn't find closing CPC for {0}, so skipping cleanup and reporting", serverInfo);
964
        return;
×
965
      }
966

967
      cleanUpResourceTimers(cpcClosed);
1✔
968

969
      if (status.isOk()) {
1✔
970
        return; // Not considered an error
1✔
971
      }
972

973
      metricReporter.reportServerFailure(1L, serverInfo.target());
1✔
974

975
      Collection<String> authoritiesForClosedCpc = getActiveAuthorities(cpcClosed);
1✔
976
      for (Map<String, ResourceSubscriber<? extends ResourceUpdate>> subscriberMap :
977
          resourceSubscribers.values()) {
1✔
978
        for (ResourceSubscriber<? extends ResourceUpdate> subscriber : subscriberMap.values()) {
1✔
979
          if (subscriber.hasResult() || !authoritiesForClosedCpc.contains(subscriber.authority)) {
1✔
980
            continue;
1✔
981
          }
982

983
          // try to fallback to lower priority control plane client
984
          if (shouldTryFallback && manageControlPlaneClient(subscriber).didFallback) {
1✔
985
            authoritiesForClosedCpc.remove(subscriber.authority);
1✔
986
            if (authoritiesForClosedCpc.isEmpty()) {
1✔
987
              return; // optimization: no need to continue once all authorities have done fallback
1✔
988
            }
989
            continue; // since we did fallback, don't consider it an error
990
          }
991

992
          subscriber.onError(status, null);
1✔
993
        }
1✔
994
      }
1✔
995
    }
1✔
996

997
  }
998

999
  private static class CpcWithFallbackState {
1000
    ControlPlaneClient cpc;
1001
    boolean didFallback;
1002

1003
    private CpcWithFallbackState(ControlPlaneClient cpc, boolean didFallback) {
1✔
1004
      this.cpc = cpc;
1✔
1005
      this.didFallback = didFallback;
1✔
1006
    }
1✔
1007
  }
1008

1009
  private Collection<String> getActiveAuthorities(ControlPlaneClient cpc) {
1010
    List<String> asList = activatedCpClients.entrySet().stream()
1✔
1011
        .filter(entry -> !entry.getValue().isEmpty()
1✔
1012
            && cpc == entry.getValue().get(entry.getValue().size() - 1))
1✔
1013
        .map(Map.Entry::getKey)
1✔
1014
        .collect(Collectors.toList());
1✔
1015

1016
    // Since this is usually used for contains, use a set when the list is large
1017
    return (asList.size() < 100) ? asList : new HashSet<>(asList);
1✔
1018
  }
1019

1020
}
STATUS · Troubleshooting · Open an Issue · Sales · Support · CAREERS · ENTERPRISE · START FREE · SCHEDULE DEMO
ANNOUNCEMENTS · TWITTER · TOS & SLA · Supported CI Services · What's a CI service? · Automated Testing

© 2025 Coveralls, Inc