#19694

Committed 14 Feb 2025 10:46PM CUT coverage: 88.585% (-0.04%) from 88.626%

Build # #19694

Build Type

push

github

Committed by

web-flow

Commit Message

okhttp:Use a locally specified value instead of Segment.SIZE in okhttp (#11899)

Switched to using 8192 which is the current value of Segment.SIZE and just have a test check that they are equal.  

The reason for doing this is that Segment.SIZE is Kotlin internal so shouldn't be used outside of its module.

Run Details

34254 of 38668 relevant lines covered (88.58%)

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 / #19694

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