#19596

Committed 19 Dec 2024 03:54PM UTC coverage: 88.598% (-0.009%) from 88.607%

Build # #19596

Build Type

push

github

Committed by

web-flow

Commit Message

Re-enable animalsniffer, fixing violations

In 61f19d707a I swapped the signatures to use the version catalog. But I
failed to preserve the `@signature` extension and it all seemed to
work... But in fact all the animalsniffer tasks were completing as
SKIPPED as they lacked signatures. The build.gradle changes in this
commit are to fix that while still using version catalog.

But while it was broken violations crept in. Most violations weren't
too important and we're not surprised went unnoticed. For example, Netty
with TLS has long required the Java 8 API
`setEndpointIdentificationAlgorithm()`, so using `Optional` in the same
code path didn't harm anything in particular. I still swapped it to
Guava's `Optional` to avoid overuse of `@IgnoreJRERequirement`.

One important violation has not been fixed and instead I've disabled the
android signature in api/build.gradle for the moment.  The violation is
in StatusException using the `fillInStackTrace` overload of Exception.
This problem [had been noticed][PR11066], but we couldn't figure out
what was going on. AnimalSniffer is now noticing this and agreeing with
the internal linter. There is still a question of why our interop tests
failed to notice this, but given they are no longer running on pre-API
level 24, that may forever be a mystery.

[PR11066]: https://github.com/grpc/grpc-java/pull/11066

Run Details

33481 of 37790 relevant lines covered (88.6%)

0.89 hits per line

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

91.94

/../xds/src/main/java/io/grpc/xds/client/ControlPlaneClient.java

/*
 * Copyright 2020 The gRPC Authors
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package io.grpc.xds.client;

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.Stopwatch;
import com.google.common.base.Supplier;
import com.google.protobuf.Any;
import com.google.rpc.Code;
import io.envoyproxy.envoy.service.discovery.v3.AggregatedDiscoveryServiceGrpc;
import io.envoyproxy.envoy.service.discovery.v3.DiscoveryRequest;
import io.envoyproxy.envoy.service.discovery.v3.DiscoveryResponse;
import io.grpc.InternalLogId;
import io.grpc.MethodDescriptor;
import io.grpc.Status;
import io.grpc.SynchronizationContext;
import io.grpc.SynchronizationContext.ScheduledHandle;
import io.grpc.internal.BackoffPolicy;
import io.grpc.xds.client.Bootstrapper.ServerInfo;
import io.grpc.xds.client.EnvoyProtoData.Node;
import io.grpc.xds.client.XdsClient.ProcessingTracker;
import io.grpc.xds.client.XdsClient.ResourceStore;
import io.grpc.xds.client.XdsClient.XdsResponseHandler;
import io.grpc.xds.client.XdsLogger.XdsLogLevel;
import io.grpc.xds.client.XdsTransportFactory.StreamingCall;
import io.grpc.xds.client.XdsTransportFactory.XdsTransport;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import javax.annotation.Nullable;

/**
 * Common base type for XdsClient implementations, which encapsulates the layer abstraction of
 * the xDS RPC stream.
 */
final class ControlPlaneClient {

  private final SynchronizationContext syncContext;
  private final InternalLogId logId;
  private final XdsLogger logger;
  private final ServerInfo serverInfo;
  private final XdsTransport xdsTransport;
  private final XdsResponseHandler xdsResponseHandler;
  private final ResourceStore resourceStore;
  private final ScheduledExecutorService timeService;
  private final BackoffPolicy.Provider backoffPolicyProvider;
  private final Stopwatch stopwatch;
  private final Node bootstrapNode;

  // Last successfully applied version_info for each resource type. Starts with empty string.
  // A version_info is used to update management server with client's most recent knowledge of
  // resources.
  private final Map<XdsResourceType<?>, String> versions = new HashMap<>();

  private boolean shutdown;
  private boolean inError;

  @Nullable
  private AdsStream adsStream;
  @Nullable
  private BackoffPolicy retryBackoffPolicy;
  @Nullable
  private ScheduledHandle rpcRetryTimer;
  private final MessagePrettyPrinter messagePrinter;

  /** An entity that manages ADS RPCs over a single channel. */
  ControlPlaneClient(
      XdsTransport xdsTransport,
      ServerInfo serverInfo,
      Node bootstrapNode,
      XdsResponseHandler xdsResponseHandler,
      ResourceStore resourceStore,
      ScheduledExecutorService
      timeService,
      SynchronizationContext syncContext,
      BackoffPolicy.Provider backoffPolicyProvider,
      Supplier<Stopwatch> stopwatchSupplier,
      MessagePrettyPrinter messagePrinter) {
    this.serverInfo = checkNotNull(serverInfo, "serverInfo");
    this.xdsTransport = checkNotNull(xdsTransport, "xdsTransport");
    this.xdsResponseHandler = checkNotNull(xdsResponseHandler, "xdsResponseHandler");
    this.resourceStore = checkNotNull(resourceStore, "resourcesSubscriber");
    this.bootstrapNode = checkNotNull(bootstrapNode, "bootstrapNode");
    this.timeService = checkNotNull(timeService, "timeService");
    this.syncContext = checkNotNull(syncContext, "syncContext");
    this.backoffPolicyProvider = checkNotNull(backoffPolicyProvider, "backoffPolicyProvider");
    this.messagePrinter = checkNotNull(messagePrinter, "messagePrinter");
    stopwatch = checkNotNull(stopwatchSupplier, "stopwatchSupplier").get();
    logId = InternalLogId.allocate("xds-client", serverInfo.target());
    logger = XdsLogger.withLogId(logId);
    logger.log(XdsLogLevel.INFO, "Created");
  }

  void shutdown() {
    syncContext.execute(new Runnable() {
      @Override
      public void run() {
        shutdown = true;
        logger.log(XdsLogLevel.INFO, "Shutting down");
        if (adsStream != null) {
          adsStream.close(Status.CANCELLED.withDescription("shutdown").asException());
        }
        if (rpcRetryTimer != null && rpcRetryTimer.isPending()) {
          rpcRetryTimer.cancel();
        }
        xdsTransport.shutdown();
      }
    });
  }

  @Override
  public String toString() {
    return logId.toString();
  }

  public ServerInfo getServerInfo() {
    return serverInfo;
  }

  /**
   * Updates the resource subscription for the given resource type.
   */
  // Must be synchronized.
  void adjustResourceSubscription(XdsResourceType<?> resourceType) {
    if (rpcRetryTimer != null && rpcRetryTimer.isPending()) {
      return;
    }
    if (adsStream == null) {
      startRpcStream();
      // when the stream becomes ready, it will send the discovery requests
      return;
    }

    // We will do the rest of the method as part of the readyHandler when the stream is ready.
    if (!isConnected()) {
      return;
    }

    Collection<String> resources = resourceStore.getSubscribedResources(serverInfo, resourceType);
    if (resources == null) {
      resources = Collections.emptyList();
    }
    adsStream.sendDiscoveryRequest(resourceType, resources);
    resourceStore.startMissingResourceTimers(resources, resourceType);

    if (resources.isEmpty()) {
      // The resource type no longer has subscribing resources; clean up references to it
      versions.remove(resourceType);
      adsStream.respNonces.remove(resourceType);
    }
  }

  /**
   * Accepts the update for the given resource type by updating the latest resource version
   * and sends an ACK request to the management server.
   */
  // Must be synchronized.
  void ackResponse(XdsResourceType<?> type, String versionInfo, String nonce) {
    versions.put(type, versionInfo);
    logger.log(XdsLogLevel.INFO, "Sending ACK for {0} update, nonce: {1}, current version: {2}",
        type.typeName(), nonce, versionInfo);
    Collection<String> resources = resourceStore.getSubscribedResources(serverInfo, type);
    if (resources == null) {
      resources = Collections.emptyList();
    }
    adsStream.sendDiscoveryRequest(type, versionInfo, resources, nonce, null);
  }

  /**
   * Rejects the update for the given resource type and sends an NACK request (request with last
   * accepted version) to the management server.
   */
  // Must be synchronized.
  void nackResponse(XdsResourceType<?> type, String nonce, String errorDetail) {
    String versionInfo = versions.getOrDefault(type, "");
    logger.log(XdsLogLevel.INFO, "Sending NACK for {0} update, nonce: {1}, current version: {2}",
        type.typeName(), nonce, versionInfo);
    Collection<String> resources = resourceStore.getSubscribedResources(serverInfo, type);
    if (resources == null) {
      resources = Collections.emptyList();
    }
    adsStream.sendDiscoveryRequest(type, versionInfo, resources, nonce, errorDetail);
  }

  // Must be synchronized.
  boolean isReady() {
    return adsStream != null && adsStream.call != null
        && adsStream.call.isReady() && !adsStream.closed;
  }

  boolean isConnected() {
    return adsStream != null && adsStream.sentInitialRequest;
  }

  /**
   * Used for identifying whether or not when getting a control plane for authority that this
   * control plane should be skipped over if there is a fallback.
   *
   * <p>Also used by metric to consider this control plane to not be "active".
   *
   * <p>A ControlPlaneClient is considered to be in error during the time from when an
   * {@link AdsStream} closed without having received a response to the time an AdsStream does
   * receive a response.
   */
  boolean isInError() {
    return inError;
  }


  /**
   * Cleans up outstanding rpcRetryTimer if present, since we are communicating.
   * If we haven't sent the initial discovery request for this RPC stream, we will delegate to
   * xdsResponseHandler (in practice XdsClientImpl) to do any initialization for a new active
   * stream such as starting timers.  We then send the initial discovery request.
   */
  // Must be synchronized.
  void readyHandler(boolean shouldSendInitialRequest) {
    if (shouldSendInitialRequest) {
      sendDiscoveryRequests();
    }
  }

  /**
   * Establishes the RPC connection by creating a new RPC stream on the given channel for
   * xDS protocol communication.
   */
  // Must be synchronized.
  private void startRpcStream() {
    checkState(adsStream == null, "Previous adsStream has not been cleared yet");

    if (rpcRetryTimer != null) {
      rpcRetryTimer.cancel();
      rpcRetryTimer = null;
    }

    adsStream = new AdsStream();
    adsStream.start();
    logger.log(XdsLogLevel.INFO, "ADS stream started");
    stopwatch.reset().start();
  }

  void sendDiscoveryRequests() {
    if (rpcRetryTimer != null && rpcRetryTimer.isPending()) {
      return;
    }

    if (adsStream == null) {
      startRpcStream();
      // when the stream becomes ready, it will send the discovery requests
      return;
    }

    if (isConnected()) {
      Set<XdsResourceType<?>> subscribedResourceTypes =
          new HashSet<>(resourceStore.getSubscribedResourceTypesWithTypeUrl().values());

      for (XdsResourceType<?> type : subscribedResourceTypes) {
        adjustResourceSubscription(type);
      }
    }
  }

  @VisibleForTesting
  public final class RpcRetryTask implements Runnable {
    @Override
    public void run() {
      logger.log(XdsLogLevel.DEBUG, "Retry timeout. Restart ADS stream {0}", logId);
      if (shutdown) {
        return;
      }

      startRpcStream();

      // handling CPC management is triggered in readyHandler
    }
  }

  @VisibleForTesting
  @Nullable
  XdsResourceType<?> fromTypeUrl(String typeUrl) {
    return resourceStore.getSubscribedResourceTypesWithTypeUrl().get(typeUrl);
  }

  private class AdsStream implements XdsTransportFactory.EventHandler<DiscoveryResponse> {
    private boolean responseReceived;
    private boolean sentInitialRequest;
    private boolean closed;
    // Response nonce for the most recently received discovery responses of each resource type.
    // Client initiated requests start response nonce with empty string.
    // Nonce in each response is echoed back in the following ACK/NACK request. It is
    // used for management server to identify which response the client is ACKing/NACking.
    // To avoid confusion, client-initiated requests will always use the nonce in
    // most recently received responses of each resource type.
    private final Map<XdsResourceType<?>, String> respNonces = new HashMap<>();
    private final StreamingCall<DiscoveryRequest, DiscoveryResponse> call;
    private final MethodDescriptor<DiscoveryRequest, DiscoveryResponse> methodDescriptor =
        AggregatedDiscoveryServiceGrpc.getStreamAggregatedResourcesMethod();

    private AdsStream() {
      this.call = xdsTransport.createStreamingCall(methodDescriptor.getFullMethodName(),
          methodDescriptor.getRequestMarshaller(), methodDescriptor.getResponseMarshaller());
    }

    void start() {
      call.start(this);
    }

    /**
     * Sends a discovery request with the given {@code versionInfo}, {@code nonce} and
     * {@code errorDetail}. Used for reacting to a specific discovery response. For
     * client-initiated discovery requests, use {@link
     * #sendDiscoveryRequest(XdsResourceType, Collection)}.
     */
    void sendDiscoveryRequest(XdsResourceType<?> type, String versionInfo,
                              Collection<String> resources, String nonce,
                              @Nullable String errorDetail) {
      DiscoveryRequest.Builder builder =
          DiscoveryRequest.newBuilder()
              .setVersionInfo(versionInfo)
              .setNode(bootstrapNode.toEnvoyProtoNode())
              .addAllResourceNames(resources)
              .setTypeUrl(type.typeUrl())
              .setResponseNonce(nonce);
      if (errorDetail != null) {
        com.google.rpc.Status error =
            com.google.rpc.Status.newBuilder()
                .setCode(Code.INVALID_ARGUMENT_VALUE)  // FIXME(chengyuanzhang): use correct code
                .setMessage(errorDetail)
                .build();
        builder.setErrorDetail(error);
      }
      DiscoveryRequest request = builder.build();
      call.sendMessage(request);
      if (logger.isLoggable(XdsLogLevel.DEBUG)) {
        logger.log(XdsLogLevel.DEBUG, "Sent DiscoveryRequest\n{0}", messagePrinter.print(request));
      }
    }

    /**
     * Sends a client-initiated discovery request.
     */
    final void sendDiscoveryRequest(XdsResourceType<?> type, Collection<String> resources) {
      logger.log(XdsLogLevel.INFO, "Sending {0} request for resources: {1}", type, resources);
      sendDiscoveryRequest(type, versions.getOrDefault(type, ""), resources,
          respNonces.getOrDefault(type, ""), null);
    }

    @Override
    public void onReady() {
      syncContext.execute(() -> {
        if (!isReady()) {
          logger.log(XdsLogLevel.DEBUG,
              "ADS stream ready handler called, but not ready {0}", logId);
          return;
        }

        logger.log(XdsLogLevel.DEBUG, "ADS stream ready {0}", logId);

        boolean hadSentInitialRequest = sentInitialRequest;
        sentInitialRequest = true;
        readyHandler(!hadSentInitialRequest);
      });
    }

    @Override
    public void onRecvMessage(DiscoveryResponse response) {
      syncContext.execute(new Runnable() {
        @Override
        public void run() {
          if (closed) {
            return;
          }
          boolean isFirstResponse = !responseReceived;
          responseReceived = true;
          inError = false;

          XdsResourceType<?> type = fromTypeUrl(response.getTypeUrl());
          if (logger.isLoggable(XdsLogLevel.DEBUG)) {
            logger.log(
                XdsLogLevel.DEBUG, "Received {0} response:\n{1}", type,
                messagePrinter.print(response));
          }
          if (type == null) {
            logger.log(
                XdsLogLevel.WARNING,
                "Ignore an unknown type of DiscoveryResponse: {0}",
                response.getTypeUrl());

            call.startRecvMessage();
            return;
          }
          handleRpcResponse(type, response.getVersionInfo(), response.getResourcesList(),
              response.getNonce(), isFirstResponse);
        }
      });
    }

    @Override
    public void onStatusReceived(final Status status) {
      syncContext.execute(() -> {
        handleRpcStreamClosed(status);
      });
    }

    final void handleRpcResponse(XdsResourceType<?> type, String versionInfo, List<Any> resources,
                                 String nonce, boolean isFirstResponse) {
      checkNotNull(type, "type");

      respNonces.put(type, nonce);
      ProcessingTracker processingTracker = new ProcessingTracker(
          () -> call.startRecvMessage(), syncContext);
      xdsResponseHandler.handleResourceResponse(type, serverInfo, versionInfo, resources, nonce,
          isFirstResponse, processingTracker);
      processingTracker.onComplete();
    }

    private void handleRpcStreamClosed(Status status) {
      if (closed) {
        return;
      }

      if (responseReceived || retryBackoffPolicy == null) {
        // Reset the backoff sequence if had received a response, or backoff sequence
        // has never been initialized.
        retryBackoffPolicy = backoffPolicyProvider.get();
      }

      // FakeClock in tests isn't thread-safe. Schedule the retry timer before notifying callbacks
      // to avoid TSAN races, since tests may wait until callbacks are called but then would run
      // concurrently with the stopwatch and schedule.

      long elapsed = stopwatch.elapsed(TimeUnit.NANOSECONDS);
      long delayNanos = Math.max(0, retryBackoffPolicy.nextBackoffNanos() - elapsed);

      rpcRetryTimer =
          syncContext.schedule(new RpcRetryTask(), delayNanos, TimeUnit.NANOSECONDS, timeService);

      Status newStatus = status;
      if (responseReceived) {
        // A closed ADS stream after a successful response is not considered an error. Servers may
        // close streams for various reasons during normal operation, such as load balancing or
        // underlying connection hitting its max connection age limit  (see gRFC A9).
        if (!status.isOk()) {
          newStatus = Status.OK;
          logger.log( XdsLogLevel.DEBUG, "ADS stream closed with error {0}: {1}. However, a "
              + "response was received, so this will not be treated as an error. Cause: {2}",
              status.getCode(), status.getDescription(), status.getCause());
        } else {
          logger.log(XdsLogLevel.DEBUG,
              "ADS stream closed by server after a response was received");
        }
      } else {
        // If the ADS stream is closed without ever having received a response from the server, then
        // the XdsClient should consider that a connectivity error (see gRFC A57).
        inError = true;
        if (status.isOk()) {
          newStatus = Status.UNAVAILABLE.withDescription(
              "ADS stream closed with OK before receiving a response");
        }
        logger.log(
            XdsLogLevel.ERROR, "ADS stream failed with status {0}: {1}. Cause: {2}",
            newStatus.getCode(), newStatus.getDescription(), newStatus.getCause());
      }

      closed = true;
      xdsResponseHandler.handleStreamClosed(newStatus, !responseReceived);
      cleanUp();
    }

    private void close(Exception error) {
      if (closed) {
        return;
      }
      closed = true;
      cleanUp();
      call.sendError(error);
    }

    private void cleanUp() {
      if (adsStream == this) {
        adsStream = null;
      }
    }
  }

  @VisibleForTesting
  static class FailingXdsTransport implements XdsTransport {
    Status error;

    public FailingXdsTransport(Status error) {
      this.error = error;
    }

    @Override
    public <ReqT, RespT> StreamingCall<ReqT, RespT>
        createStreamingCall(String fullMethodName,
                            MethodDescriptor.Marshaller<ReqT> reqMarshaller,
                            MethodDescriptor.Marshaller<RespT> respMarshaller) {
      return new FailingXdsStreamingCall<>();
    }

    @Override
    public void shutdown() {
      // no-op
    }

    private class FailingXdsStreamingCall<ReqT, RespT> implements StreamingCall<ReqT, RespT> {

      @Override
      public void start(XdsTransportFactory.EventHandler<RespT> eventHandler) {
        eventHandler.onStatusReceived(error);
      }

      @Override
      public void sendMessage(ReqT message) {
        // no-op
      }

      @Override
      public void startRecvMessage() {
        // no-op
      }

      @Override
      public void sendError(Exception e) {
        // no-op
      }

      @Override
      public boolean isReady() {
        return false;
      }
    }
  }

}

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.checkNotNull;
20	import static com.google.common.base.Preconditions.checkState;
21
22	import com.google.common.annotations.VisibleForTesting;
23	import com.google.common.base.Stopwatch;
24	import com.google.common.base.Supplier;
25	import com.google.protobuf.Any;
26	import com.google.rpc.Code;
27	import io.envoyproxy.envoy.service.discovery.v3.AggregatedDiscoveryServiceGrpc;
28	import io.envoyproxy.envoy.service.discovery.v3.DiscoveryRequest;
29	import io.envoyproxy.envoy.service.discovery.v3.DiscoveryResponse;
30	import io.grpc.InternalLogId;
31	import io.grpc.MethodDescriptor;
32	import io.grpc.Status;
33	import io.grpc.SynchronizationContext;
34	import io.grpc.SynchronizationContext.ScheduledHandle;
35	import io.grpc.internal.BackoffPolicy;
36	import io.grpc.xds.client.Bootstrapper.ServerInfo;
37	import io.grpc.xds.client.EnvoyProtoData.Node;
38	import io.grpc.xds.client.XdsClient.ProcessingTracker;
39	import io.grpc.xds.client.XdsClient.ResourceStore;
40	import io.grpc.xds.client.XdsClient.XdsResponseHandler;
41	import io.grpc.xds.client.XdsLogger.XdsLogLevel;
42	import io.grpc.xds.client.XdsTransportFactory.StreamingCall;
43	import io.grpc.xds.client.XdsTransportFactory.XdsTransport;
44	import java.util.Collection;
45	import java.util.Collections;
46	import java.util.HashMap;
47	import java.util.HashSet;
48	import java.util.List;
49	import java.util.Map;
50	import java.util.Set;
51	import java.util.concurrent.ScheduledExecutorService;
52	import java.util.concurrent.TimeUnit;
53	import javax.annotation.Nullable;
54
55	/**
56	* Common base type for XdsClient implementations, which encapsulates the layer abstraction of
57	* the xDS RPC stream.
58	*/
59	final class ControlPlaneClient {
60
61	private final SynchronizationContext syncContext;
62	private final InternalLogId logId;
63	private final XdsLogger logger;
64	private final ServerInfo serverInfo;
65	private final XdsTransport xdsTransport;
66	private final XdsResponseHandler xdsResponseHandler;
67	private final ResourceStore resourceStore;
68	private final ScheduledExecutorService timeService;
69	private final BackoffPolicy.Provider backoffPolicyProvider;
70	private final Stopwatch stopwatch;
71	private final Node bootstrapNode;
72
73	// Last successfully applied version_info for each resource type. Starts with empty string.
74	// A version_info is used to update management server with client's most recent knowledge of
75	// resources.
76	private final Map<XdsResourceType<?>, String> versions = new HashMap<>();	1✔
77
78	private boolean shutdown;
79	private boolean inError;
80
81	@Nullable
82	private AdsStream adsStream;
83	@Nullable
84	private BackoffPolicy retryBackoffPolicy;
85	@Nullable
86	private ScheduledHandle rpcRetryTimer;
87	private final MessagePrettyPrinter messagePrinter;
88
89	/** An entity that manages ADS RPCs over a single channel. */
90	ControlPlaneClient(
91	XdsTransport xdsTransport,
92	ServerInfo serverInfo,
93	Node bootstrapNode,
94	XdsResponseHandler xdsResponseHandler,
95	ResourceStore resourceStore,
96	ScheduledExecutorService
97	timeService,
98	SynchronizationContext syncContext,
99	BackoffPolicy.Provider backoffPolicyProvider,
100	Supplier<Stopwatch> stopwatchSupplier,
101	MessagePrettyPrinter messagePrinter) {	1✔
102	this.serverInfo = checkNotNull(serverInfo, "serverInfo");	1✔
103	this.xdsTransport = checkNotNull(xdsTransport, "xdsTransport");	1✔
104	this.xdsResponseHandler = checkNotNull(xdsResponseHandler, "xdsResponseHandler");	1✔
105	this.resourceStore = checkNotNull(resourceStore, "resourcesSubscriber");	1✔
106	this.bootstrapNode = checkNotNull(bootstrapNode, "bootstrapNode");	1✔
107	this.timeService = checkNotNull(timeService, "timeService");	1✔
108	this.syncContext = checkNotNull(syncContext, "syncContext");	1✔
109	this.backoffPolicyProvider = checkNotNull(backoffPolicyProvider, "backoffPolicyProvider");	1✔
110	this.messagePrinter = checkNotNull(messagePrinter, "messagePrinter");	1✔
111	stopwatch = checkNotNull(stopwatchSupplier, "stopwatchSupplier").get();	1✔
112	logId = InternalLogId.allocate("xds-client", serverInfo.target());	1✔
113	logger = XdsLogger.withLogId(logId);	1✔
114	logger.log(XdsLogLevel.INFO, "Created");	1✔
115	}	1✔
116
117	void shutdown() {
118	syncContext.execute(new Runnable() {	1✔
119	@Override
120	public void run() {
121	shutdown = true;	1✔
122	logger.log(XdsLogLevel.INFO, "Shutting down");	1✔
123	if (adsStream != null) {	1✔
124	adsStream.close(Status.CANCELLED.withDescription("shutdown").asException());	1✔
125	}
126	if (rpcRetryTimer != null && rpcRetryTimer.isPending()) {	1✔
127	rpcRetryTimer.cancel();	1✔
128	}
129	xdsTransport.shutdown();	1✔
130	}	1✔
131	});
132	}	1✔
133
134	@Override
135	public String toString() {
136	return logId.toString();	×
137	}
138
139	public ServerInfo getServerInfo() {
140	return serverInfo;	1✔
141	}
142
143	/**
144	* Updates the resource subscription for the given resource type.
145	*/
146	// Must be synchronized.
147	void adjustResourceSubscription(XdsResourceType<?> resourceType) {
148	if (rpcRetryTimer != null && rpcRetryTimer.isPending()) {	1✔
149	return;	1✔
150	}
151	if (adsStream == null) {	1✔
152	startRpcStream();	×
153	// when the stream becomes ready, it will send the discovery requests
154	return;	×
155	}
156
157	// We will do the rest of the method as part of the readyHandler when the stream is ready.
158	if (!isConnected()) {	1✔
159	return;	1✔
160	}
161
162	Collection<String> resources = resourceStore.getSubscribedResources(serverInfo, resourceType);	1✔
163	if (resources == null) {	1✔
164	resources = Collections.emptyList();	1✔
165	}
166	adsStream.sendDiscoveryRequest(resourceType, resources);	1✔
167	resourceStore.startMissingResourceTimers(resources, resourceType);	1✔
168
169	if (resources.isEmpty()) {	1✔
170	// The resource type no longer has subscribing resources; clean up references to it
171	versions.remove(resourceType);	1✔
172	adsStream.respNonces.remove(resourceType);	1✔
173	}
174	}	1✔
175
176	/**
177	* Accepts the update for the given resource type by updating the latest resource version
178	* and sends an ACK request to the management server.
179	*/
180	// Must be synchronized.
181	void ackResponse(XdsResourceType<?> type, String versionInfo, String nonce) {
182	versions.put(type, versionInfo);	1✔
183	logger.log(XdsLogLevel.INFO, "Sending ACK for {0} update, nonce: {1}, current version: {2}",	1✔
184	type.typeName(), nonce, versionInfo);	1✔
185	Collection<String> resources = resourceStore.getSubscribedResources(serverInfo, type);	1✔
186	if (resources == null) {	1✔
187	resources = Collections.emptyList();	1✔
188	}
189	adsStream.sendDiscoveryRequest(type, versionInfo, resources, nonce, null);	1✔
190	}	1✔
191
192	/**
193	* Rejects the update for the given resource type and sends an NACK request (request with last
194	* accepted version) to the management server.
195	*/
196	// Must be synchronized.
197	void nackResponse(XdsResourceType<?> type, String nonce, String errorDetail) {
198	String versionInfo = versions.getOrDefault(type, "");	1✔
199	logger.log(XdsLogLevel.INFO, "Sending NACK for {0} update, nonce: {1}, current version: {2}",	1✔
200	type.typeName(), nonce, versionInfo);	1✔
201	Collection<String> resources = resourceStore.getSubscribedResources(serverInfo, type);	1✔
202	if (resources == null) {	1✔
203	resources = Collections.emptyList();	×
204	}
205	adsStream.sendDiscoveryRequest(type, versionInfo, resources, nonce, errorDetail);	1✔
206	}	1✔
207
208	// Must be synchronized.
209	boolean isReady() {
210	return adsStream != null && adsStream.call != null	1✔
211	&& adsStream.call.isReady() && !adsStream.closed;	1✔
212	}
213
214	boolean isConnected() {
215	return adsStream != null && adsStream.sentInitialRequest;	1✔
216	}
217
218	/**
219	* Used for identifying whether or not when getting a control plane for authority that this
220	* control plane should be skipped over if there is a fallback.
221	*
222	* <p>Also used by metric to consider this control plane to not be "active".
223	*
224	* <p>A ControlPlaneClient is considered to be in error during the time from when an
225	* {@link AdsStream} closed without having received a response to the time an AdsStream does
226	* receive a response.
227	*/
228	boolean isInError() {
229	return inError;	1✔
230	}
231
232
233	/**
234	* Cleans up outstanding rpcRetryTimer if present, since we are communicating.
235	* If we haven't sent the initial discovery request for this RPC stream, we will delegate to
236	* xdsResponseHandler (in practice XdsClientImpl) to do any initialization for a new active
237	* stream such as starting timers. We then send the initial discovery request.
238	*/
239	// Must be synchronized.
240	void readyHandler(boolean shouldSendInitialRequest) {
241	if (shouldSendInitialRequest) {	1✔
242	sendDiscoveryRequests();	1✔
243	}
244	}	1✔
245
246	/**
247	* Establishes the RPC connection by creating a new RPC stream on the given channel for
248	* xDS protocol communication.
249	*/
250	// Must be synchronized.
251	private void startRpcStream() {
252	checkState(adsStream == null, "Previous adsStream has not been cleared yet");	1✔
253
254	if (rpcRetryTimer != null) {	1✔
255	rpcRetryTimer.cancel();	1✔
256	rpcRetryTimer = null;	1✔
257	}
258
259	adsStream = new AdsStream();	1✔
260	adsStream.start();	1✔
261	logger.log(XdsLogLevel.INFO, "ADS stream started");	1✔
262	stopwatch.reset().start();	1✔
263	}	1✔
264
265	void sendDiscoveryRequests() {
266	if (rpcRetryTimer != null && rpcRetryTimer.isPending()) {	1✔
267	return;	×
268	}
269
270	if (adsStream == null) {	1✔
271	startRpcStream();	1✔
272	// when the stream becomes ready, it will send the discovery requests
273	return;	1✔
274	}
275
276	if (isConnected()) {	1✔
277	Set<XdsResourceType<?>> subscribedResourceTypes =	1✔
278	new HashSet<>(resourceStore.getSubscribedResourceTypesWithTypeUrl().values());	1✔
279
280	for (XdsResourceType<?> type : subscribedResourceTypes) {	1✔
281	adjustResourceSubscription(type);	1✔
282	}	1✔
283	}
284	}	1✔
285
286	@VisibleForTesting
287	public final class RpcRetryTask implements Runnable {	1✔
288	@Override
289	public void run() {
290	logger.log(XdsLogLevel.DEBUG, "Retry timeout. Restart ADS stream {0}", logId);	1✔
291	if (shutdown) {	1✔
292	return;	×
293	}
294
295	startRpcStream();	1✔
296
297	// handling CPC management is triggered in readyHandler
298	}	1✔
299	}
300
301	@VisibleForTesting
302	@Nullable
303	XdsResourceType<?> fromTypeUrl(String typeUrl) {
304	return resourceStore.getSubscribedResourceTypesWithTypeUrl().get(typeUrl);	1✔
305	}
306
307	private class AdsStream implements XdsTransportFactory.EventHandler<DiscoveryResponse> {
308	private boolean responseReceived;
309	private boolean sentInitialRequest;
310	private boolean closed;
311	// Response nonce for the most recently received discovery responses of each resource type.
312	// Client initiated requests start response nonce with empty string.
313	// Nonce in each response is echoed back in the following ACK/NACK request. It is
314	// used for management server to identify which response the client is ACKing/NACking.
315	// To avoid confusion, client-initiated requests will always use the nonce in
316	// most recently received responses of each resource type.
317	private final Map<XdsResourceType<?>, String> respNonces = new HashMap<>();	1✔
318	private final StreamingCall<DiscoveryRequest, DiscoveryResponse> call;
319	private final MethodDescriptor<DiscoveryRequest, DiscoveryResponse> methodDescriptor =	1✔
320	AggregatedDiscoveryServiceGrpc.getStreamAggregatedResourcesMethod();	1✔
321
322	private AdsStream() {	1✔
323	this.call = xdsTransport.createStreamingCall(methodDescriptor.getFullMethodName(),	1✔
324	methodDescriptor.getRequestMarshaller(), methodDescriptor.getResponseMarshaller());	1✔
325	}	1✔
326
327	void start() {
328	call.start(this);	1✔
329	}	1✔
330
331	/**
332	* Sends a discovery request with the given {@code versionInfo}, {@code nonce} and
333	* {@code errorDetail}. Used for reacting to a specific discovery response. For
334	* client-initiated discovery requests, use {@link
335	* #sendDiscoveryRequest(XdsResourceType, Collection)}.
336	*/
337	void sendDiscoveryRequest(XdsResourceType<?> type, String versionInfo,
338	Collection<String> resources, String nonce,
339	@Nullable String errorDetail) {
340	DiscoveryRequest.Builder builder =
341	DiscoveryRequest.newBuilder()	1✔
342	.setVersionInfo(versionInfo)	1✔
343	.setNode(bootstrapNode.toEnvoyProtoNode())	1✔
344	.addAllResourceNames(resources)	1✔
345	.setTypeUrl(type.typeUrl())	1✔
346	.setResponseNonce(nonce);	1✔
347	if (errorDetail != null) {	1✔
348	com.google.rpc.Status error =
349	com.google.rpc.Status.newBuilder()	1✔
350	.setCode(Code.INVALID_ARGUMENT_VALUE) // FIXME(chengyuanzhang): use correct code	1✔
351	.setMessage(errorDetail)	1✔
352	.build();	1✔
353	builder.setErrorDetail(error);	1✔
354	}
355	DiscoveryRequest request = builder.build();	1✔
356	call.sendMessage(request);	1✔
357	if (logger.isLoggable(XdsLogLevel.DEBUG)) {	1✔
358	logger.log(XdsLogLevel.DEBUG, "Sent DiscoveryRequest\n{0}", messagePrinter.print(request));	×
359	}
360	}	1✔
361
362	/**
363	* Sends a client-initiated discovery request.
364	*/
365	final void sendDiscoveryRequest(XdsResourceType<?> type, Collection<String> resources) {
366	logger.log(XdsLogLevel.INFO, "Sending {0} request for resources: {1}", type, resources);	1✔
367	sendDiscoveryRequest(type, versions.getOrDefault(type, ""), resources,	1✔
368	respNonces.getOrDefault(type, ""), null);	1✔
369	}	1✔
370
371	@Override
372	public void onReady() {
373	syncContext.execute(() -> {	1✔
374	if (!isReady()) {	1✔
375	logger.log(XdsLogLevel.DEBUG,	×
376	"ADS stream ready handler called, but not ready {0}", logId);	×
377	return;	×
378	}
379
380	logger.log(XdsLogLevel.DEBUG, "ADS stream ready {0}", logId);	1✔
381
382	boolean hadSentInitialRequest = sentInitialRequest;	1✔
383	sentInitialRequest = true;	1✔
384	readyHandler(!hadSentInitialRequest);	1✔
385	});	1✔
386	}	1✔
387
388	@Override
389	public void onRecvMessage(DiscoveryResponse response) {
390	syncContext.execute(new Runnable() {	1✔
391	@Override
392	public void run() {
393	if (closed) {	1✔
394	return;	1✔
395	}
396	boolean isFirstResponse = !responseReceived;	1✔
397	responseReceived = true;	1✔
398	inError = false;	1✔
399
400	XdsResourceType<?> type = fromTypeUrl(response.getTypeUrl());	1✔
401	if (logger.isLoggable(XdsLogLevel.DEBUG)) {	1✔
402	logger.log(	×
403	XdsLogLevel.DEBUG, "Received {0} response:\n{1}", type,
404	messagePrinter.print(response));	×
405	}
406	if (type == null) {	1✔
407	logger.log(	1✔
408	XdsLogLevel.WARNING,
409	"Ignore an unknown type of DiscoveryResponse: {0}",
410	response.getTypeUrl());	1✔
411
412	call.startRecvMessage();	1✔
413	return;	1✔
414	}
415	handleRpcResponse(type, response.getVersionInfo(), response.getResourcesList(),	1✔
416	response.getNonce(), isFirstResponse);	1✔
417	}	1✔
418	});
419	}	1✔
420
421	@Override
422	public void onStatusReceived(final Status status) {
423	syncContext.execute(() -> {	1✔
424	handleRpcStreamClosed(status);	1✔
425	});	1✔
426	}	1✔
427
428	final void handleRpcResponse(XdsResourceType<?> type, String versionInfo, List<Any> resources,
429	String nonce, boolean isFirstResponse) {
430	checkNotNull(type, "type");	1✔
431
432	respNonces.put(type, nonce);	1✔
433	ProcessingTracker processingTracker = new ProcessingTracker(	1✔
434	() -> call.startRecvMessage(), syncContext);	1✔
435	xdsResponseHandler.handleResourceResponse(type, serverInfo, versionInfo, resources, nonce,	1✔
436	isFirstResponse, processingTracker);
437	processingTracker.onComplete();	1✔
438	}	1✔
439
440	private void handleRpcStreamClosed(Status status) {
441	if (closed) {	1✔
442	return;	1✔
443	}
444
445	if (responseReceived \|\| retryBackoffPolicy == null) {	1✔
446	// Reset the backoff sequence if had received a response, or backoff sequence
447	// has never been initialized.
448	retryBackoffPolicy = backoffPolicyProvider.get();	1✔
449	}
450
451	// FakeClock in tests isn't thread-safe. Schedule the retry timer before notifying callbacks
452	// to avoid TSAN races, since tests may wait until callbacks are called but then would run
453	// concurrently with the stopwatch and schedule.
454
455	long elapsed = stopwatch.elapsed(TimeUnit.NANOSECONDS);	1✔
456	long delayNanos = Math.max(0, retryBackoffPolicy.nextBackoffNanos() - elapsed);	1✔
457
458	rpcRetryTimer =	1✔
459	syncContext.schedule(new RpcRetryTask(), delayNanos, TimeUnit.NANOSECONDS, timeService);	1✔
460
461	Status newStatus = status;	1✔
462	if (responseReceived) {	1✔
463	// A closed ADS stream after a successful response is not considered an error. Servers may
464	// close streams for various reasons during normal operation, such as load balancing or
465	// underlying connection hitting its max connection age limit (see gRFC A9).
466	if (!status.isOk()) {	1✔
467	newStatus = Status.OK;	1✔
468	logger.log( XdsLogLevel.DEBUG, "ADS stream closed with error {0}: {1}. However, a "	1✔
469	+ "response was received, so this will not be treated as an error. Cause: {2}",
470	status.getCode(), status.getDescription(), status.getCause());	1✔
471	} else {
472	logger.log(XdsLogLevel.DEBUG,	1✔
473	"ADS stream closed by server after a response was received");
474	}
475	} else {
476	// If the ADS stream is closed without ever having received a response from the server, then
477	// the XdsClient should consider that a connectivity error (see gRFC A57).
478	inError = true;	1✔
479	if (status.isOk()) {	1✔
480	newStatus = Status.UNAVAILABLE.withDescription(	1✔
481	"ADS stream closed with OK before receiving a response");
482	}
483	logger.log(	1✔
484	XdsLogLevel.ERROR, "ADS stream failed with status {0}: {1}. Cause: {2}",
485	newStatus.getCode(), newStatus.getDescription(), newStatus.getCause());	1✔
486	}
487
488	closed = true;	1✔
489	xdsResponseHandler.handleStreamClosed(newStatus, !responseReceived);	1✔
490	cleanUp();	1✔
491	}	1✔
492
493	private void close(Exception error) {
494	if (closed) {	1✔
495	return;	×
496	}
497	closed = true;	1✔
498	cleanUp();	1✔
499	call.sendError(error);	1✔
500	}	1✔
501
502	private void cleanUp() {
503	if (adsStream == this) {	1✔
504	adsStream = null;	1✔
505	}
506	}	1✔
507	}
508
509	@VisibleForTesting
510	static class FailingXdsTransport implements XdsTransport {
511	Status error;
512
513	public FailingXdsTransport(Status error) {	1✔
514	this.error = error;	1✔
515	}	1✔
516
517	@Override
518	public <ReqT, RespT> StreamingCall<ReqT, RespT>
519	createStreamingCall(String fullMethodName,
520	MethodDescriptor.Marshaller<ReqT> reqMarshaller,
521	MethodDescriptor.Marshaller<RespT> respMarshaller) {
522	return new FailingXdsStreamingCall<>();	1✔
523	}
524
525	@Override
526	public void shutdown() {
527	// no-op
528	}	1✔
529
530	private class FailingXdsStreamingCall<ReqT, RespT> implements StreamingCall<ReqT, RespT> {	1✔
531
532	@Override
533	public void start(XdsTransportFactory.EventHandler<RespT> eventHandler) {
534	eventHandler.onStatusReceived(error);	1✔
535	}	1✔
536
537	@Override
538	public void sendMessage(ReqT message) {
539	// no-op
540	}	×
541
542	@Override
543	public void startRecvMessage() {
544	// no-op
545	}	×
546
547	@Override
548	public void sendError(Exception e) {
549	// no-op
550	}	×
551
552	@Override
553	public boolean isReady() {
554	return false;	×
555	}
556	}
557	}
558
559	}

grpc / grpc-java / #19596

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