Merge branch 'grpc:master' into master

Author: tharakadesilva

core/src/main/java/io/grpc/internal/AbstractClientStream.java

@@ -455,10 +455,10 @@ private void closeListener(
       if (!listenerClosed) {
         listenerClosed = true;
         statsTraceCtx.streamClosed(status);
-        listener().closed(status, rpcProgress, trailers);
         if (getTransportTracer() != null) {
           getTransportTracer().reportStreamClosed(status.isOk());
         }
+        listener().closed(status, rpcProgress, trailers);
       }
     }
   }

interop-testing/src/test/java/io/grpc/testing/integration/RetryTest.java

@@ -18,7 +18,10 @@
 
 import static com.google.common.truth.Truth.assertThat;
 import static java.util.concurrent.TimeUnit.SECONDS;
+import static org.junit.Assert.assertNotNull;
+import static org.mockito.AdditionalAnswers.delegatesTo;
 import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.never;
 import static org.mockito.Mockito.timeout;
 import static org.mockito.Mockito.verify;
@@ -78,8 +81,6 @@
 import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.junit.runners.JUnit4;
-import org.mockito.ArgumentCaptor;
-import org.mockito.Mock;
 import org.mockito.junit.MockitoJUnit;
 import org.mockito.junit.MockitoRule;
 
@@ -103,8 +104,11 @@ public class RetryTest {
   @Rule
   public final GrpcCleanupRule cleanupRule = new GrpcCleanupRule();
   private final FakeClock fakeClock = new FakeClock();
-  @Mock
-  private ClientCall.Listener<Integer> mockCallListener;
+  private TestListener testCallListener = new TestListener();
+  @SuppressWarnings("unchecked")
+  private ClientCall.Listener<Integer> mockCallListener =
+      mock(ClientCall.Listener.class, delegatesTo(testCallListener));
+
   private CountDownLatch backoffLatch = new CountDownLatch(1);
   private final EventLoopGroup group = new DefaultEventLoopGroup() {
     @SuppressWarnings("FutureReturnValueIgnored")
@@ -245,7 +249,9 @@ private void assertInboundWireSizeRecorded(long length) throws Exception {
   private void assertRpcStatusRecorded(
       Status.Code code, long roundtripLatencyMs, long outboundMessages) throws Exception {
     MetricsRecord record = clientStatsRecorder.pollRecord(7, SECONDS);
+    assertNotNull(record);
     TagValue statusTag = record.tags.get(RpcMeasureConstants.GRPC_CLIENT_STATUS);
+    assertNotNull(statusTag);
     assertThat(statusTag.asString()).isEqualTo(code.toString());
     assertThat(record.getMetricAsLongOrFail(DeprecatedCensusConstants.RPC_CLIENT_FINISHED_COUNT))
         .isEqualTo(1);
@@ -295,14 +301,14 @@ public void retryUntilBufferLimitExceeded() throws Exception {
     verify(mockCallListener, never()).onClose(any(Status.class), any(Metadata.class));
     // send one more message, should exceed buffer limit
     call.sendMessage(message);
+
     // let attempt fail
+    testCallListener.clear();
     serverCall.close(
         Status.UNAVAILABLE.withDescription("2nd attempt failed"),
         new Metadata());
     // no more retry
-    ArgumentCaptor<Status> statusCaptor = ArgumentCaptor.forClass(Status.class);
-    verify(mockCallListener, timeout(5000)).onClose(statusCaptor.capture(), any(Metadata.class));
-    assertThat(statusCaptor.getValue().getDescription()).contains("2nd attempt failed");
+    testCallListener.verifyDescription("2nd attempt failed", 5000);
   }
 
   @Test
@@ -534,4 +540,26 @@ public ClientStreamTracer newClientStreamTracer(StreamInfo info, Metadata header
     assertRpcStatusRecorded(Code.INVALID_ARGUMENT, 0, 0);
     assertRetryStatsRecorded(0, 1, 0);
   }
+
+  private static class TestListener extends ClientCall.Listener<Integer> {
+    Status status = null;
+    private CountDownLatch closeLatch = new CountDownLatch(1);
+
+    @Override
+    public void onClose(Status status, Metadata trailers) {
+      this.status = status;
+      closeLatch.countDown();
+    }
+
+    void clear() {
+      status = null;
+      closeLatch = new CountDownLatch(1);
+    }
+
+    void verifyDescription(String description, long timeoutMs) throws InterruptedException {
+      closeLatch.await(timeoutMs, TimeUnit.MILLISECONDS);
+      assertNotNull(status);
+      assertThat(status.getDescription()).contains(description);
+    }
+  }
 }

xds/src/main/java/io/grpc/xds/XdsClient.java

@@ -48,6 +48,7 @@
  * protocols (e.g., LDS, RDS, VHDS, CDS and EDS) over a single channel. Watch-based interfaces
  * are provided for each set of data needed by gRPC.
  */
+@ExperimentalApi("https://github.com/grpc/grpc-java/issues/10862")
 public abstract class XdsClient {
 
   static boolean isResourceNameValid(String resourceName, String typeUrl) {

xds/src/main/java/io/grpc/xds/XdsListenerResource.java

@@ -130,7 +130,8 @@ private LdsUpdate processServerSideListener(Listener proto, Args args)
     if (args.bootstrapInfo != null && args.bootstrapInfo.certProviders() != null) {
       certProviderInstances = args.bootstrapInfo.certProviders().keySet();
     }
-    return LdsUpdate.forTcpListener(parseServerSideListener(proto, args.tlsContextManager,
+    return LdsUpdate.forTcpListener(parseServerSideListener(proto,
+        (TlsContextManager) args.securityConfig,
         filterRegistry, certProviderInstances));
   }
 

xds/src/main/java/io/grpc/xds/XdsResourceType.java

@@ -85,12 +85,13 @@ public abstract class XdsResourceType<T extends ResourceUpdate> {
   // the resources that need an update.
   protected abstract boolean isFullStateOfTheWorld();
 
+  @ExperimentalApi("https://github.com/grpc/grpc-java/issues/10847")
   public static class Args {
     final ServerInfo serverInfo;
     final String versionInfo;
     final String nonce;
     final Bootstrapper.BootstrapInfo bootstrapInfo;
-    final TlsContextManager tlsContextManager;
+    final Object securityConfig;
     // Management server is required to always send newly requested resources, even if they
     // may have been sent previously (proactively). Thus, client does not need to cache
     // unrequested resources.
@@ -99,17 +100,18 @@ public static class Args {
 
     public Args(ServerInfo serverInfo, String versionInfo, String nonce,
                 Bootstrapper.BootstrapInfo bootstrapInfo,
-                TlsContextManager tlsContextManager,
+                Object securityConfig,
                 @Nullable Set<String> subscribedResources) {
       this.serverInfo = serverInfo;
       this.versionInfo = versionInfo;
       this.nonce = nonce;
       this.bootstrapInfo = bootstrapInfo;
-      this.tlsContextManager = tlsContextManager;
+      this.securityConfig = securityConfig;
       this.subscribedResources = subscribedResources;
     }
   }
 
+  @ExperimentalApi("https://github.com/grpc/grpc-java/issues/10847")
   public static final class ResourceInvalidException extends Exception {
     private static final long serialVersionUID = 0L;
 

xds/third_party/envoy/import.sh

@@ -17,7 +17,7 @@
 
 set -e
 # import VERSION from the google internal copybara_version.txt for Envoy
-VERSION=0478eba2a495027bf6ac8e787c42e2f5b9eb553b
+VERSION=147e6b9523d8d2ae0d9d2205254d6e633644c6fe
 DOWNLOAD_URL="https://github.com/envoyproxy/envoy/archive/${VERSION}.tar.gz"
 DOWNLOAD_BASE_DIR="envoy-${VERSION}"
 SOURCE_PROTO_BASE_DIR="${DOWNLOAD_BASE_DIR}/api"
@@ -79,6 +79,7 @@ envoy/config/core/v3/event_service_config.proto
 envoy/config/core/v3/extension.proto
 envoy/config/core/v3/grpc_service.proto
 envoy/config/core/v3/health_check.proto
+envoy/config/core/v3/http_service.proto
 envoy/config/core/v3/http_uri.proto
 envoy/config/core/v3/protocol.proto
 envoy/config/core/v3/proxy_protocol.proto
@@ -124,6 +125,7 @@ envoy/config/trace/v3/opentelemetry.proto
 envoy/config/trace/v3/service.proto
 envoy/config/trace/v3/trace.proto
 envoy/config/trace/v3/zipkin.proto
+envoy/data/accesslog/v3/accesslog.proto
 envoy/extensions/clusters/aggregate/v3/cluster.proto
 envoy/extensions/filters/common/fault/v3/fault.proto
 envoy/extensions/filters/http/fault/v3/fault.proto

xds/third_party/envoy/src/main/proto/envoy/admin/v3/config_dump.proto

@@ -33,6 +33,7 @@ message ConfigDump {
   // * ``bootstrap``: :ref:`BootstrapConfigDump <envoy_v3_api_msg_admin.v3.BootstrapConfigDump>`
   // * ``clusters``: :ref:`ClustersConfigDump <envoy_v3_api_msg_admin.v3.ClustersConfigDump>`
   // * ``ecds_filter_http``: :ref:`EcdsConfigDump <envoy_v3_api_msg_admin.v3.EcdsConfigDump>`
+  // * ``ecds_filter_quic_listener``: :ref:`EcdsConfigDump <envoy_v3_api_msg_admin.v3.EcdsConfigDump>`
   // * ``ecds_filter_tcp_listener``: :ref:`EcdsConfigDump <envoy_v3_api_msg_admin.v3.EcdsConfigDump>`
   // * ``endpoints``:  :ref:`EndpointsConfigDump <envoy_v3_api_msg_admin.v3.EndpointsConfigDump>`
   // * ``listeners``: :ref:`ListenersConfigDump <envoy_v3_api_msg_admin.v3.ListenersConfigDump>`

xds/third_party/envoy/src/main/proto/envoy/api/v2/scoped_route.proto

@@ -39,7 +39,7 @@ option (udpa.annotations.file_status).package_version_status = FROZEN;
 //       fragments:
 //         - header_value_extractor:
 //             name: X-Route-Selector
-//             element_separator: ,
+//             element_separator: ","
 //             element:
 //               separator: =
 //               key: vip

xds/third_party/envoy/src/main/proto/envoy/config/accesslog/v3/accesslog.proto

@@ -4,6 +4,7 @@ package envoy.config.accesslog.v3;
 
 import "envoy/config/core/v3/base.proto";
 import "envoy/config/route/v3/route_components.proto";
+import "envoy/data/accesslog/v3/accesslog.proto";
 import "envoy/type/matcher/v3/metadata.proto";
 import "envoy/type/v3/percent.proto";
 
@@ -43,7 +44,7 @@ message AccessLog {
   }
 }
 
-// [#next-free-field: 13]
+// [#next-free-field: 14]
 message AccessLogFilter {
   option (udpa.annotations.versioning).previous_message_type =
       "envoy.config.filter.accesslog.v2.AccessLogFilter";
@@ -87,6 +88,9 @@ message AccessLogFilter {
 
     // Metadata Filter
     MetadataFilter metadata_filter = 12;
+
+    // Log Type Filter
+    LogTypeFilter log_type_filter = 13;
   }
 }
 
@@ -250,6 +254,8 @@ message ResponseFlagFilter {
         in: "UPE"
         in: "NC"
         in: "OM"
+        in: "DF"
+        in: "DO"
       }
     }
   }];
@@ -310,6 +316,17 @@ message MetadataFilter {
   google.protobuf.BoolValue match_if_key_not_found = 2;
 }
 
+// Filters based on access log type.
+message LogTypeFilter {
+  // Logs only records which their type is one of the types defined in this field.
+  repeated data.accesslog.v3.AccessLogType types = 1
+      [(validate.rules).repeated = {items {enum {defined_only: true}}}];
+
+  // If this field is set to true, the filter will instead block all records
+  // with a access log type in types field, and allow all other records.
+  bool exclude = 2;
+}
+
 // Extension filter is statically registered at runtime.
 message ExtensionFilter {
   option (udpa.annotations.versioning).previous_message_type =

xds/third_party/envoy/src/main/proto/envoy/config/bootstrap/v3/bootstrap.proto

@@ -41,7 +41,7 @@ option (udpa.annotations.file_status).package_version_status = ACTIVE;
 // <config_overview_bootstrap>` for more detail.
 
 // Bootstrap :ref:`configuration overview <config_overview_bootstrap>`.
-// [#next-free-field: 38]
+// [#next-free-field: 41]
 message Bootstrap {
   option (udpa.annotations.versioning).previous_message_type =
       "envoy.config.bootstrap.v2.Bootstrap";
@@ -101,6 +101,48 @@ message Bootstrap {
     core.v3.ApiConfigSource ads_config = 3;
   }
 
+  message ApplicationLogConfig {
+    message LogFormat {
+      oneof log_format {
+        option (validate.required) = true;
+
+        // Flush application logs in JSON format. The configured JSON struct can
+        // support all the format flags specified in the :option:`--log-format`
+        // command line options section, except for the ``%v`` and ``%_`` flags.
+        google.protobuf.Struct json_format = 1;
+
+        // Flush application log in a format defined by a string. The text format
+        // can support all the format flags specified in the :option:`--log-format`
+        // command line option section.
+        string text_format = 2;
+      }
+    }
+
+    // Optional field to set the application logs format. If this field is set, it will override
+    // the default log format. Setting both this field and :option:`--log-format` command line
+    // option is not allowed, and will cause a bootstrap error.
+    LogFormat log_format = 1;
+  }
+
+  message DeferredStatOptions {
+    // When the flag is enabled, Envoy will lazily initialize a subset of the stats (see below).
+    // This will save memory and CPU cycles when creating the objects that own these stats, if those
+    // stats are never referenced throughout the lifetime of the process. However, it will incur additional
+    // memory overhead for these objects, and a small increase of CPU usage when a at least one of the stats
+    // is updated for the first time.
+    // Groups of stats that will be lazily initialized:
+    // - Cluster traffic stats: a subgroup of the :ref:`cluster statistics <config_cluster_manager_cluster_stats>`
+    // that are used when requests are routed to the cluster.
+    bool enable_deferred_creation_stats = 1;
+  }
+
+  message GrpcAsyncClientManagerConfig {
+    // Optional field to set the expiration time for the cached gRPC client object.
+    // The minimal value is 5s and the default is 50s.
+    google.protobuf.Duration max_cached_entry_idle_duration = 1
+        [(validate.rules).duration = {gte {seconds: 5}}];
+  }
+
   reserved 10, 11;
 
   reserved "runtime";
@@ -163,6 +205,9 @@ message Bootstrap {
   // Optional set of stats sinks.
   repeated metrics.v3.StatsSink stats_sinks = 6;
 
+  // Options to control behaviors of deferred creation compatible stats.
+  DeferredStatOptions deferred_stat_options = 39;
+
   // Configuration for internal processing of stats.
   metrics.v3.StatsConfig stats_config = 13;
 
@@ -360,6 +405,12 @@ message Bootstrap {
   // Envoy only supports ListenerManager for this field and Envoy Mobile
   // supports ApiListenerManager.
   core.v3.TypedExtensionConfig listener_manager = 37;
+
+  // Optional application log configuration.
+  ApplicationLogConfig application_log_config = 38;
+
+  // Optional gRPC async manager config.
+  GrpcAsyncClientManagerConfig grpc_async_client_manager_config = 40;
 }
 
 // Administration interface :ref:`operations documentation
@@ -397,6 +448,7 @@ message Admin {
 }
 
 // Cluster manager :ref:`architecture overview <arch_overview_cluster_manager>`.
+// [#next-free-field: 6]
 message ClusterManager {
   option (udpa.annotations.versioning).previous_message_type =
       "envoy.config.bootstrap.v2.ClusterManager";
@@ -437,6 +489,11 @@ message ClusterManager {
   // <envoy_v3_api_field_config.core.v3.ApiConfigSource.api_type>` :ref:`GRPC
   // <envoy_v3_api_enum_value_config.core.v3.ApiConfigSource.ApiType.GRPC>`.
   core.v3.ApiConfigSource load_stats_config = 4;
+
+  // Whether the ClusterManager will create clusters on the worker threads
+  // inline during requests. This will save memory and CPU cycles in cases where
+  // there are lots of inactive clusters and > 1 worker thread.
+  bool enable_deferred_cluster_creation = 5;
 }
 
 // Allows you to specify different watchdog configs for different subsystems.