#19688

Committed 13 Feb 2025 08:41PM CUT coverage: 88.613% (+0.008%) from 88.605%

Build # #19688

Build Type

push

github

Committed by

ejona86

Commit Message

Upgrade netty-tcnative to 2.0.70

Run Details

34248 of 38649 relevant lines covered (88.61%)

0.89 hits per line

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

92.38

/../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, except
      // for nonces. If the resource type becomes used again the control plane can ignore requests
      // for old/missing nonces. Old type's nonces are dropped when the ADS stream is restarted.
      versions.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 URL.
    // 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. Nonces are never deleted from the
    // map; nonces are only discarded once the stream closes because xds_protocol says "the
    // management server should not send a DiscoveryResponse for any DiscoveryRequest that has a
    // stale nonce."
    private final Map<String, 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.typeUrl(), ""), 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;
          respNonces.put(response.getTypeUrl(), response.getNonce());

          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");

      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();
        stopwatch.reset();
      }

      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());
      }

      close(newStatus.asException());

      // 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);

      xdsResponseHandler.handleStreamClosed(newStatus, !responseReceived);
    }

    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, except
171	// for nonces. If the resource type becomes used again the control plane can ignore requests
172	// for old/missing nonces. Old type's nonces are dropped when the ADS stream is restarted.
173	versions.remove(resourceType);	1✔
174	}
175	}	1✔
176
177	/**
178	* Accepts the update for the given resource type by updating the latest resource version
179	* and sends an ACK request to the management server.
180	*/
181	// Must be synchronized.
182	void ackResponse(XdsResourceType<?> type, String versionInfo, String nonce) {
183	versions.put(type, versionInfo);	1✔
184	logger.log(XdsLogLevel.INFO, "Sending ACK for {0} update, nonce: {1}, current version: {2}",	1✔
185	type.typeName(), nonce, versionInfo);	1✔
186	Collection<String> resources = resourceStore.getSubscribedResources(serverInfo, type);	1✔
187	if (resources == null) {	1✔
188	resources = Collections.emptyList();	1✔
189	}
190	adsStream.sendDiscoveryRequest(type, versionInfo, resources, nonce, null);	1✔
191	}	1✔
192
193	/**
194	* Rejects the update for the given resource type and sends an NACK request (request with last
195	* accepted version) to the management server.
196	*/
197	// Must be synchronized.
198	void nackResponse(XdsResourceType<?> type, String nonce, String errorDetail) {
199	String versionInfo = versions.getOrDefault(type, "");	1✔
200	logger.log(XdsLogLevel.INFO, "Sending NACK for {0} update, nonce: {1}, current version: {2}",	1✔
201	type.typeName(), nonce, versionInfo);	1✔
202	Collection<String> resources = resourceStore.getSubscribedResources(serverInfo, type);	1✔
203	if (resources == null) {	1✔
204	resources = Collections.emptyList();	×
205	}
206	adsStream.sendDiscoveryRequest(type, versionInfo, resources, nonce, errorDetail);	1✔
207	}	1✔
208
209	// Must be synchronized.
210	boolean isReady() {
211	return adsStream != null && adsStream.call != null	1✔
212	&& adsStream.call.isReady() && !adsStream.closed;	1✔
213	}
214
215	boolean isConnected() {
216	return adsStream != null && adsStream.sentInitialRequest;	1✔
217	}
218
219	/**
220	* Used for identifying whether or not when getting a control plane for authority that this
221	* control plane should be skipped over if there is a fallback.
222	*
223	* <p>Also used by metric to consider this control plane to not be "active".
224	*
225	* <p>A ControlPlaneClient is considered to be in error during the time from when an
226	* {@link AdsStream} closed without having received a response to the time an AdsStream does
227	* receive a response.
228	*/
229	boolean isInError() {
230	return inError;	1✔
231	}
232
233
234	/**
235	* Cleans up outstanding rpcRetryTimer if present, since we are communicating.
236	* If we haven't sent the initial discovery request for this RPC stream, we will delegate to
237	* xdsResponseHandler (in practice XdsClientImpl) to do any initialization for a new active
238	* stream such as starting timers. We then send the initial discovery request.
239	*/
240	// Must be synchronized.
241	void readyHandler(boolean shouldSendInitialRequest) {
242	if (shouldSendInitialRequest) {	1✔
243	sendDiscoveryRequests();	1✔
244	}
245	}	1✔
246
247	/**
248	* Establishes the RPC connection by creating a new RPC stream on the given channel for
249	* xDS protocol communication.
250	*/
251	// Must be synchronized.
252	private void startRpcStream() {
253	checkState(adsStream == null, "Previous adsStream has not been cleared yet");	1✔
254
255	if (rpcRetryTimer != null) {	1✔
256	rpcRetryTimer.cancel();	1✔
257	rpcRetryTimer = null;	1✔
258	}
259
260	adsStream = new AdsStream();	1✔
261	adsStream.start();	1✔
262	logger.log(XdsLogLevel.INFO, "ADS stream started");	1✔
263	stopwatch.reset().start();	1✔
264	}	1✔
265
266	void sendDiscoveryRequests() {
267	if (rpcRetryTimer != null && rpcRetryTimer.isPending()) {	1✔
268	return;	×
269	}
270
271	if (adsStream == null) {	1✔
272	startRpcStream();	1✔
273	// when the stream becomes ready, it will send the discovery requests
274	return;	1✔
275	}
276
277	if (isConnected()) {	1✔
278	Set<XdsResourceType<?>> subscribedResourceTypes =	1✔
279	new HashSet<>(resourceStore.getSubscribedResourceTypesWithTypeUrl().values());	1✔
280
281	for (XdsResourceType<?> type : subscribedResourceTypes) {	1✔
282	adjustResourceSubscription(type);	1✔
283	}	1✔
284	}
285	}	1✔
286
287	@VisibleForTesting
288	public final class RpcRetryTask implements Runnable {	1✔
289	@Override
290	public void run() {
291	logger.log(XdsLogLevel.DEBUG, "Retry timeout. Restart ADS stream {0}", logId);	1✔
292	if (shutdown) {	1✔
293	return;	×
294	}
295
296	startRpcStream();	1✔
297
298	// handling CPC management is triggered in readyHandler
299	}	1✔
300	}
301
302	@VisibleForTesting
303	@Nullable
304	XdsResourceType<?> fromTypeUrl(String typeUrl) {
305	return resourceStore.getSubscribedResourceTypesWithTypeUrl().get(typeUrl);	1✔
306	}
307
308	private class AdsStream implements XdsTransportFactory.EventHandler<DiscoveryResponse> {
309	private boolean responseReceived;
310	private boolean sentInitialRequest;
311	private boolean closed;
312	// Response nonce for the most recently received discovery responses of each resource type URL.
313	// Client initiated requests start response nonce with empty string.
314	// Nonce in each response is echoed back in the following ACK/NACK request. It is
315	// used for management server to identify which response the client is ACKing/NACking.
316	// To avoid confusion, client-initiated requests will always use the nonce in
317	// most recently received responses of each resource type. Nonces are never deleted from the
318	// map; nonces are only discarded once the stream closes because xds_protocol says "the
319	// management server should not send a DiscoveryResponse for any DiscoveryRequest that has a
320	// stale nonce."
321	private final Map<String, String> respNonces = new HashMap<>();	1✔
322	private final StreamingCall<DiscoveryRequest, DiscoveryResponse> call;
323	private final MethodDescriptor<DiscoveryRequest, DiscoveryResponse> methodDescriptor =	1✔
324	AggregatedDiscoveryServiceGrpc.getStreamAggregatedResourcesMethod();	1✔
325
326	private AdsStream() {	1✔
327	this.call = xdsTransport.createStreamingCall(methodDescriptor.getFullMethodName(),	1✔
328	methodDescriptor.getRequestMarshaller(), methodDescriptor.getResponseMarshaller());	1✔
329	}	1✔
330
331	void start() {
332	call.start(this);	1✔
333	}	1✔
334
335	/**
336	* Sends a discovery request with the given {@code versionInfo}, {@code nonce} and
337	* {@code errorDetail}. Used for reacting to a specific discovery response. For
338	* client-initiated discovery requests, use {@link
339	* #sendDiscoveryRequest(XdsResourceType, Collection)}.
340	*/
341	void sendDiscoveryRequest(XdsResourceType<?> type, String versionInfo,
342	Collection<String> resources, String nonce,
343	@Nullable String errorDetail) {
344	DiscoveryRequest.Builder builder =
345	DiscoveryRequest.newBuilder()	1✔
346	.setVersionInfo(versionInfo)	1✔
347	.setNode(bootstrapNode.toEnvoyProtoNode())	1✔
348	.addAllResourceNames(resources)	1✔
349	.setTypeUrl(type.typeUrl())	1✔
350	.setResponseNonce(nonce);	1✔
351	if (errorDetail != null) {	1✔
352	com.google.rpc.Status error =
353	com.google.rpc.Status.newBuilder()	1✔
354	.setCode(Code.INVALID_ARGUMENT_VALUE) // FIXME(chengyuanzhang): use correct code	1✔
355	.setMessage(errorDetail)	1✔
356	.build();	1✔
357	builder.setErrorDetail(error);	1✔
358	}
359	DiscoveryRequest request = builder.build();	1✔
360	call.sendMessage(request);	1✔
361	if (logger.isLoggable(XdsLogLevel.DEBUG)) {	1✔
362	logger.log(XdsLogLevel.DEBUG, "Sent DiscoveryRequest\n{0}", messagePrinter.print(request));	×
363	}
364	}	1✔
365
366	/**
367	* Sends a client-initiated discovery request.
368	*/
369	final void sendDiscoveryRequest(XdsResourceType<?> type, Collection<String> resources) {
370	logger.log(XdsLogLevel.INFO, "Sending {0} request for resources: {1}", type, resources);	1✔
371	sendDiscoveryRequest(type, versions.getOrDefault(type, ""), resources,	1✔
372	respNonces.getOrDefault(type.typeUrl(), ""), null);	1✔
373	}	1✔
374
375	@Override
376	public void onReady() {
377	syncContext.execute(() -> {	1✔
378	if (!isReady()) {	1✔
379	logger.log(XdsLogLevel.DEBUG,	×
380	"ADS stream ready handler called, but not ready {0}", logId);	×
381	return;	×
382	}
383
384	logger.log(XdsLogLevel.DEBUG, "ADS stream ready {0}", logId);	1✔
385
386	boolean hadSentInitialRequest = sentInitialRequest;	1✔
387	sentInitialRequest = true;	1✔
388	readyHandler(!hadSentInitialRequest);	1✔
389	});	1✔
390	}	1✔
391
392	@Override
393	public void onRecvMessage(DiscoveryResponse response) {
394	syncContext.execute(new Runnable() {	1✔
395	@Override
396	public void run() {
397	if (closed) {	1✔
398	return;	1✔
399	}
400	boolean isFirstResponse = !responseReceived;	1✔
401	responseReceived = true;	1✔
402	inError = false;	1✔
403	respNonces.put(response.getTypeUrl(), response.getNonce());	1✔
404
405	XdsResourceType<?> type = fromTypeUrl(response.getTypeUrl());	1✔
406	if (logger.isLoggable(XdsLogLevel.DEBUG)) {	1✔
407	logger.log(	×
408	XdsLogLevel.DEBUG, "Received {0} response:\n{1}", type,
409	messagePrinter.print(response));	×
410	}
411	if (type == null) {	1✔
412	logger.log(	1✔
413	XdsLogLevel.WARNING,
414	"Ignore an unknown type of DiscoveryResponse: {0}",
415	response.getTypeUrl());	1✔
416
417	call.startRecvMessage();	1✔
418	return;	1✔
419	}
420	handleRpcResponse(type, response.getVersionInfo(), response.getResourcesList(),	1✔
421	response.getNonce(), isFirstResponse);	1✔
422	}	1✔
423	});
424	}	1✔
425
426	@Override
427	public void onStatusReceived(final Status status) {
428	syncContext.execute(() -> {	1✔
429	handleRpcStreamClosed(status);	1✔
430	});	1✔
431	}	1✔
432
433	final void handleRpcResponse(XdsResourceType<?> type, String versionInfo, List<Any> resources,
434	String nonce, boolean isFirstResponse) {
435	checkNotNull(type, "type");	1✔
436
437	ProcessingTracker processingTracker = new ProcessingTracker(	1✔
438	() -> call.startRecvMessage(), syncContext);	1✔
439	xdsResponseHandler.handleResourceResponse(type, serverInfo, versionInfo, resources, nonce,	1✔
440	isFirstResponse, processingTracker);
441	processingTracker.onComplete();	1✔
442	}	1✔
443
444	private void handleRpcStreamClosed(Status status) {
445	if (closed) {	1✔
446	return;	1✔
447	}
448
449	if (responseReceived \|\| retryBackoffPolicy == null) {	1✔
450	// Reset the backoff sequence if had received a response, or backoff sequence
451	// has never been initialized.
452	retryBackoffPolicy = backoffPolicyProvider.get();	1✔
453	stopwatch.reset();	1✔
454	}
455
456	Status newStatus = status;	1✔
457	if (responseReceived) {	1✔
458	// A closed ADS stream after a successful response is not considered an error. Servers may
459	// close streams for various reasons during normal operation, such as load balancing or
460	// underlying connection hitting its max connection age limit (see gRFC A9).
461	if (!status.isOk()) {	1✔
462	newStatus = Status.OK;	1✔
463	logger.log( XdsLogLevel.DEBUG, "ADS stream closed with error {0}: {1}. However, a "	1✔
464	+ "response was received, so this will not be treated as an error. Cause: {2}",
465	status.getCode(), status.getDescription(), status.getCause());	1✔
466	} else {
467	logger.log(XdsLogLevel.DEBUG,	1✔
468	"ADS stream closed by server after a response was received");
469	}
470	} else {
471	// If the ADS stream is closed without ever having received a response from the server, then
472	// the XdsClient should consider that a connectivity error (see gRFC A57).
473	inError = true;	1✔
474	if (status.isOk()) {	1✔
475	newStatus = Status.UNAVAILABLE.withDescription(	1✔
476	"ADS stream closed with OK before receiving a response");
477	}
478	logger.log(	1✔
479	XdsLogLevel.ERROR, "ADS stream failed with status {0}: {1}. Cause: {2}",
480	newStatus.getCode(), newStatus.getDescription(), newStatus.getCause());	1✔
481	}
482
483	close(newStatus.asException());	1✔
484
485	// FakeClock in tests isn't thread-safe. Schedule the retry timer before notifying callbacks
486	// to avoid TSAN races, since tests may wait until callbacks are called but then would run
487	// concurrently with the stopwatch and schedule.
488	long elapsed = stopwatch.elapsed(TimeUnit.NANOSECONDS);	1✔
489	long delayNanos = Math.max(0, retryBackoffPolicy.nextBackoffNanos() - elapsed);	1✔
490	rpcRetryTimer =	1✔
491	syncContext.schedule(new RpcRetryTask(), delayNanos, TimeUnit.NANOSECONDS, timeService);	1✔
492
493	xdsResponseHandler.handleStreamClosed(newStatus, !responseReceived);	1✔
494	}	1✔
495
496	private void close(Exception error) {
497	if (closed) {	1✔
498	return;	×
499	}
500	closed = true;	1✔
501	cleanUp();	1✔
502	call.sendError(error);	1✔
503	}	1✔
504
505	private void cleanUp() {
506	if (adsStream == this) {	1✔
507	adsStream = null;	1✔
508	}
509	}	1✔
510	}
511
512	@VisibleForTesting
513	static class FailingXdsTransport implements XdsTransport {
514	Status error;
515
516	public FailingXdsTransport(Status error) {	1✔
517	this.error = error;	1✔
518	}	1✔
519
520	@Override
521	public <ReqT, RespT> StreamingCall<ReqT, RespT>
522	createStreamingCall(String fullMethodName,
523	MethodDescriptor.Marshaller<ReqT> reqMarshaller,
524	MethodDescriptor.Marshaller<RespT> respMarshaller) {
525	return new FailingXdsStreamingCall<>();	1✔
526	}
527
528	@Override
529	public void shutdown() {
530	// no-op
531	}	1✔
532
533	private class FailingXdsStreamingCall<ReqT, RespT> implements StreamingCall<ReqT, RespT> {	1✔
534
535	@Override
536	public void start(XdsTransportFactory.EventHandler<RespT> eventHandler) {
537	eventHandler.onStatusReceived(error);	1✔
538	}	1✔
539
540	@Override
541	public void sendMessage(ReqT message) {
542	// no-op
543	}	×
544
545	@Override
546	public void startRecvMessage() {
547	// no-op
548	}	×
549
550	@Override
551	public void sendError(Exception e) {
552	// no-op
553	}	1✔
554
555	@Override
556	public boolean isReady() {
557	return false;	×
558	}
559	}
560	}
561
562	}

grpc / grpc-java / #19688

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