xDS LDS parsing changes: NACK on use_original_dst (#25687)

* xDS LDS parsing changes: NACK on use_original_dst

* Reviewer comments

* Unused variable

src/core/ext/xds/xds_api.cc

@@ -2037,6 +2037,14 @@ grpc_error* LdsResponseParseServer(
       AddressParse(envoy_config_listener_v3_Listener_address(listener),
                    &lds_update->address);
   if (error != GRPC_ERROR_NONE) return error;
+  const auto* use_original_dst =
+      envoy_config_listener_v3_Listener_use_original_dst(listener);
+  if (use_original_dst != nullptr) {
+    if (google_protobuf_BoolValue_value(use_original_dst)) {
+      return GRPC_ERROR_CREATE_FROM_STATIC_STRING(
+          "Field \'use_original_dst\' is not supported.");
+    }
+  }
   // TODO(yashykt): As part of this, we'll need to refactor the code to process
   // the HttpConnectionManager config so that it is shared with the client-side
   // parsing.

src/proto/grpc/testing/xds/v3/listener.proto

@@ -22,6 +22,7 @@ import "src/proto/grpc/testing/xds/v3/address.proto";
 import "src/proto/grpc/testing/xds/v3/base.proto";
 
 import "google/protobuf/any.proto";
+import "google/protobuf/wrappers.proto";
 
 // [#protodoc-title: Listener configuration]
 // Listener :ref:`configuration overview <config_listeners>`
@@ -89,7 +90,7 @@ message FilterChain {
   // connections established with the listener. Order matters as the filters are
   // processed sequentially as connection events happen. Note: If the filter
   // list is empty, the connection will close by default.
-  repeated Filter filters = 3;  
+  repeated Filter filters = 3; 
 
   // Optional custom transport socket implementation to use for downstream connections.
   // To setup TLS, set a transport socket with name `tls` and
@@ -120,6 +121,13 @@ message Listener {
   // :ref:`FAQ entry <faq_how_to_setup_sni>`.
   repeated FilterChain filter_chains = 3;
 
+  // If a connection is redirected using *iptables*, the port on which the proxy
+  // receives it might be different from the original destination address. When this flag is set to
+  // true, the listener hands off redirected connections to the listener associated with the
+  // original destination address. If there is no listener associated with the original destination
+  // address, the connection is handled by the listener that receives it. Defaults to false.
+  google.protobuf.BoolValue use_original_dst = 4;  
+
   // Used to represent an API listener, which is used in non-proxy clients. The type of API
   // exposed to the non-proxy application depends on the type of API listener.
   // When this field is set, no other field except for :ref:`name<envoy_api_field_config.listener.v3.Listener.name>`

test/cpp/end2end/xds_end2end_test.cc

@@ -7373,7 +7373,6 @@ TEST_P(XdsEnabledServerTest, Basic) {
       HttpConnectionManager());
   balancers_[0]->ads_service()->SetLdsResource(listener);
   WaitForBackend(0);
-  CheckRpcSendOk();
 }
 
 TEST_P(XdsEnabledServerTest, BadLdsUpdateNoApiListenerNorAddress) {
@@ -7384,7 +7383,10 @@ TEST_P(XdsEnabledServerTest, BadLdsUpdateNoApiListenerNorAddress) {
   listener.add_filter_chains()->add_filters()->mutable_typed_config()->PackFrom(
       HttpConnectionManager());
   balancers_[0]->ads_service()->SetLdsResource(listener);
-  CheckRpcSendFailure(1, RpcOptions().set_wait_for_ready(true));
+  do {
+    CheckRpcSendFailure();
+  } while (balancers_[0]->ads_service()->lds_response_state().state ==
+           AdsServiceImpl::ResponseState::SENT);
   const auto response_state =
       balancers_[0]->ads_service()->lds_response_state();
   EXPECT_EQ(response_state.state, AdsServiceImpl::ResponseState::NACKED);
@@ -7406,11 +7408,12 @@ TEST_P(XdsEnabledServerTest, BadLdsUpdateBothApiListenerAndAddress) {
   auto* filter_chain = listener.add_filter_chains();
   filter_chain->add_filters()->mutable_typed_config()->PackFrom(
       HttpConnectionManager());
-  auto* transport_socket = filter_chain->mutable_transport_socket();
-  transport_socket->set_name("envoy.transport_sockets.tls");
   listener.mutable_api_listener();
   balancers_[0]->ads_service()->SetLdsResource(listener);
-  CheckRpcSendFailure(1, RpcOptions().set_wait_for_ready(true));
+  do {
+    CheckRpcSendFailure();
+  } while (balancers_[0]->ads_service()->lds_response_state().state ==
+           AdsServiceImpl::ResponseState::SENT);
   const auto response_state =
       balancers_[0]->ads_service()->lds_response_state();
   EXPECT_EQ(response_state.state, AdsServiceImpl::ResponseState::NACKED);
@@ -7429,12 +7432,12 @@ TEST_P(XdsEnabledServerTest, UnsupportedL4Filter) {
       ipv6_only_ ? "::1" : "127.0.0.1");
   listener.mutable_address()->mutable_socket_address()->set_port_value(
       backends_[0]->port());
-  auto* filter_chain = listener.add_filter_chains();
+  listener.add_filter_chains()->add_filters()->mutable_typed_config()->PackFrom(default_listener_ /* any proto object other than HttpConnectionManager */);
-  filter_chain->add_filters()->mutable_typed_config()->PackFrom(default_listener_ /* any proto object other than HttpConnectionManager */);
-  auto* transport_socket = filter_chain->mutable_transport_socket();
-  transport_socket->set_name("envoy.transport_sockets.tls");
   balancers_[0]->ads_service()->SetLdsResource(listener);
-  CheckRpcSendFailure(1, RpcOptions().set_wait_for_ready(true));
+  do {
+    CheckRpcSendFailure();
+  } while (balancers_[0]->ads_service()->lds_response_state().state ==
+           AdsServiceImpl::ResponseState::SENT);
   const auto response_state =
       balancers_[0]->ads_service()->lds_response_state();
   EXPECT_EQ(response_state.state, AdsServiceImpl::ResponseState::NACKED);
@@ -7465,7 +7468,10 @@ TEST_P(XdsEnabledServerTest, UnsupportedHttpFilter) {
       absl::StrCat("grpc/server?xds.resource.listening_address=[::1]:",
                    backends_[0]->port()));
   balancers_[0]->ads_service()->SetLdsResource(listener);
-  CheckRpcSendFailure(1, RpcOptions().set_wait_for_ready(true));
+  do {
+    CheckRpcSendFailure();
+  } while (balancers_[0]->ads_service()->lds_response_state().state ==
+           AdsServiceImpl::ResponseState::SENT);
   const auto response_state =
       balancers_[0]->ads_service()->lds_response_state();
   EXPECT_EQ(response_state.state, AdsServiceImpl::ResponseState::NACKED);
@@ -7498,7 +7504,10 @@ TEST_P(XdsEnabledServerTest, HttpFilterNotSupportedOnServer) {
       absl::StrCat("grpc/server?xds.resource.listening_address=[::1]:",
                    backends_[0]->port()));
   balancers_[0]->ads_service()->SetLdsResource(listener);
-  CheckRpcSendFailure(1, RpcOptions().set_wait_for_ready(true));
+  do {
+    CheckRpcSendFailure();
+  } while (balancers_[0]->ads_service()->lds_response_state().state ==
+           AdsServiceImpl::ResponseState::SENT);
   const auto response_state =
       balancers_[0]->ads_service()->lds_response_state();
   EXPECT_EQ(response_state.state, AdsServiceImpl::ResponseState::NACKED);
@@ -7555,19 +7564,39 @@ TEST_P(XdsEnabledServerTest, ListenerAddressMismatch) {
       HttpConnectionManager());
   balancers_[0]->ads_service()->SetLdsResource(listener);
   WaitForBackend(0);
-  CheckRpcSendOk();
   // Set a different listening address in the LDS update
   listener.mutable_address()->mutable_socket_address()->set_address(
       "192.168.1.1");
   balancers_[0]->ads_service()->SetLdsResource(listener);
-  bool rpc_failed = false;
+  backends_[0]->notifier()->WaitOnServingStatusChange(
-  for (int i = 0; i < 100; ++i) {
+      absl::StrCat(ipv6_only_ ? "[::1]:" : "127.0.0.1:", backends_[0]->port()),
-    if (!SendRpc().ok()) {
+      grpc::StatusCode::FAILED_PRECONDITION);
-      rpc_failed = true;
+}
-      break;
+
-    }
+TEST_P(XdsEnabledServerTest, UseOriginalDstNotSupported) {
-  }
+  Listener listener;
-  EXPECT_TRUE(rpc_failed);
+  listener.set_name(
+      absl::StrCat("grpc/server?xds.resource.listening_address=",
+                   ipv6_only_ ? "[::1]:" : "127.0.0.1:", backends_[0]->port()));
+  balancers_[0]->ads_service()->SetLdsResource(listener);
+  listener.mutable_address()->mutable_socket_address()->set_address(
+      ipv6_only_ ? "::1" : "127.0.0.1");
+  listener.mutable_address()->mutable_socket_address()->set_port_value(
+      backends_[0]->port());
+  listener.add_filter_chains()->add_filters()->mutable_typed_config()->PackFrom(
+      HttpConnectionManager());
+  listener.mutable_use_original_dst()->set_value(true);
+  balancers_[0]->ads_service()->SetLdsResource(listener);
+  do {
+    CheckRpcSendFailure();
+  } while (balancers_[0]->ads_service()->lds_response_state().state ==
+           AdsServiceImpl::ResponseState::SENT);
+  const auto response_state =
+      balancers_[0]->ads_service()->lds_response_state();
+  EXPECT_EQ(response_state.state, AdsServiceImpl::ResponseState::NACKED);
+  EXPECT_THAT(
+      response_state.error_message,
+      ::testing::HasSubstr("Field \'use_original_dst\' is not supported."));
 }
 
 class XdsServerSecurityTest : public XdsEnd2endTest {
@@ -8082,7 +8111,6 @@ class XdsEnabledServerStatusNotificationTest : public XdsServerSecurityTest {
   void SetValidLdsUpdate() { SetLdsUpdate("", "", "", "", false); }
 
   void SetInvalidLdsUpdate() {
-    // Set LDS update without root provider instance.
     Listener listener;
     listener.set_name(absl::StrCat(
         "grpc/server?xds.resource.listening_address=",
@@ -8091,9 +8119,14 @@ class XdsEnabledServerStatusNotificationTest : public XdsServerSecurityTest {
     socket_address->set_address(ipv6_only_ ? "::1" : "127.0.0.1");
     socket_address->set_port_value(backends_[0]->port());
     auto* filter_chain = listener.add_filter_chains();
+    filter_chain->add_filters()->mutable_typed_config()->PackFrom(
+        HttpConnectionManager());
     auto* transport_socket = filter_chain->mutable_transport_socket();
     transport_socket->set_name("envoy.transport_sockets.tls");
     DownstreamTlsContext downstream_tls_context;
+    downstream_tls_context.mutable_common_tls_context()
+        ->mutable_tls_certificate_certificate_provider_instance()
+        ->set_instance_name("unknown");
     transport_socket->mutable_typed_config()->PackFrom(downstream_tls_context);
     balancers_[0]->ads_service()->SetLdsResource(listener);
   }
@@ -8131,15 +8164,10 @@ TEST_P(XdsEnabledServerStatusNotificationTest, ErrorUpdateWhenAlreadyServing) {
   SendRpc([this]() { return CreateInsecureChannel(); }, {}, {});
   // Invalid update does not lead to a change in the serving status.
   SetInvalidLdsUpdate();
-  constexpr int kRetryCount = 100;
+  do {
-  auto response_state = balancers_[0]->ads_service()->lds_response_state();
-  for (int i = 0; i < kRetryCount &&
-                  response_state.state != AdsServiceImpl::ResponseState::NACKED;
-       i++) {
     SendRpc([this]() { return CreateInsecureChannel(); }, {}, {});
-    response_state = balancers_[0]->ads_service()->lds_response_state();
+  } while (balancers_[0]->ads_service()->lds_response_state().state ==
-  }
+           AdsServiceImpl::ResponseState::SENT);
-  EXPECT_EQ(response_state.state, AdsServiceImpl::ResponseState::NACKED);
   backends_[0]->notifier()->WaitOnServingStatusChange(
       absl::StrCat(ipv6_only_ ? "[::1]:" : "127.0.0.1:", backends_[0]->port()),
       grpc::StatusCode::OK);