Support mixed v2/v3 xDS request/responses

source/common/config/api_type_oracle.cc

@@ -31,5 +31,26 @@ ApiTypeOracle::getEarlierVersionMessageTypeName(const std::string& message_type)
  }
  return absl::nullopt;
 }
+
+const absl::optional<std::string> ApiTypeOracle::getEarlierTypeUrl(const std::string& type_url) {
+ char delimeter = '/';
+ size_t type_name_start = type_url.size();
+ for (size_t i = 0; i < type_url.size(); i++) {
+ if (type_url[i] == delimeter) {
+ type_name_start = i;
+ }
+ }
+ if (type_name_start == type_url.size()) {
+ return {};
+ }
+ std::string message_type = type_url.substr(type_name_start + 1);
+ absl::optional<std::string> old_message_type =
+ ApiTypeOracle::getEarlierVersionMessageTypeName(message_type);
+ if (old_message_type) {
+ return "type.googleapis.com/" + *old_message_type;
+ }
+ return {};
+}
+
 } // namespace Config
 } // namespace Envoy

source/common/config/api_type_oracle.h

@@ -22,6 +22,8 @@ class ApiTypeOracle {
 
  static const absl::optional<std::string>
  getEarlierVersionMessageTypeName(const std::string& message_type);
+
+ static const absl::optional<std::string> getEarlierTypeUrl(const std::string& type_url);
 };
 
 } // namespace Config

source/common/config/grpc_mux_impl.cc

@@ -116,17 +116,25 @@ ScopedResume GrpcMuxImpl::pause(const std::vector<std::string> type_urls) {
 void GrpcMuxImpl::onDiscoveryResponse(
  std::unique_ptr<envoy::service::discovery::v3::DiscoveryResponse>&& message,
  ControlPlaneStats& control_plane_stats) {
- const std::string& type_url = message->type_url();
+ std::string type_url = message->type_url();
  ENVOY_LOG(debug, "Received gRPC message for {} at version {}", type_url, message->version_info());
  if (message->has_control_plane()) {
  control_plane_stats.identifier_.set(message->control_plane().identifier());
  }
+ // If this type url is not watched, try older version of type url.
  if (api_state_.count(type_url) == 0) {
- ENVOY_LOG(warn, "Ignoring the message for type URL {} as it has no current subscribers.",
- type_url);
- // TODO(yuval-k): This should never happen. consider dropping the stream as this is a
- // protocol violation
- return;
+ absl::optional<std::string> old_type_url =
+ ApiTypeOracle::getEarlierTypeUrl(message->type_url());
+ if (old_type_url && api_state_.count(*old_type_url)) {
+ type_url = *old_type_url;
+ ENVOY_LOG(debug, "v3 {} converted to v2 {}.", message->type_url(), *old_type_url);
+ } else {
+ // TODO(yuval-k): This should never happen. consider dropping the stream as this is a
+ // protocol violation
+ ENVOY_LOG(warn, "Ignoring the message for type URL {} as it has no current subscribers.",
+ type_url);
+ return;
+ }
  }
  if (api_state_[type_url].watches_.empty()) {
  // update the nonce as we are processing this response.
@@ -164,10 +172,10 @@ void GrpcMuxImpl::onDiscoveryResponse(
  OpaqueResourceDecoder& resource_decoder =
  api_state_[type_url].watches_.front()->resource_decoder_;
  for (const auto& resource : message->resources()) {
- if (type_url != resource.type_url()) {
+ if (message->type_url() != resource.type_url()) {
  throw EnvoyException(
  fmt::format("{} does not match the message-wide type URL {} in DiscoveryResponse {}",
- resource.type_url(), type_url, message->DebugString()));
+ resource.type_url(), message->type_url(), message->DebugString()));
  }
  resources.emplace_back(
  new DecodedResourceImpl(resource_decoder, resource, message->version_info()));

source/common/config/new_grpc_mux_impl.cc

@@ -50,6 +50,14 @@ void NewGrpcMuxImpl::onDiscoveryResponse(
  ENVOY_LOG(debug, "Received DeltaDiscoveryResponse for {} at version {}", message->type_url(),
  message->system_version_info());
  auto sub = subscriptions_.find(message->type_url());
+ // If this type url is not watched, try older version type url.
+ if (sub == subscriptions_.end()) {
+ absl::optional<std::string> old_type_url =
+ ApiTypeOracle::getEarlierTypeUrl(message->type_url());
+ if (old_type_url) {
+ sub = subscriptions_.find(*old_type_url);
+ }
+ }
  if (sub == subscriptions_.end()) {
  ENVOY_LOG(warn,
  "Dropping received DeltaDiscoveryResponse (with version {}) for non-existent "

test/common/config/grpc_mux_impl_test.cc

@@ -722,6 +722,94 @@ TEST_F(GrpcMuxImplTest, BadLocalInfoEmptyNodeName) {
  "ads: node 'id' and 'cluster' are required. Set it either in 'node' config or via "
  "--service-node and --service-cluster options.");
 }
+
+// Send discovery request with v2 resource type_url, receive discovery response with v3 resource
+// type_url.
+TEST_F(GrpcMuxImplTest, WatchV2ResourceV3) {
+ setup();
+ InSequence s;
+ TestUtility::TestOpaqueResourceDecoderImpl<envoy::config::endpoint::v3::ClusterLoadAssignment>
+ resource_decoder("cluster_name");
+ const std::string& v2_type_url = Config::TypeUrl::get().ClusterLoadAssignment;
+ const std::string& v3_type_url =
+ "type.googleapis.com/envoy.config.endpoint.v3.ClusterLoadAssignment";
+ NiceMock<MockSubscriptionCallbacks> foo_callbacks;
+ auto foo_sub = grpc_mux_->addWatch(v2_type_url, {"x", "y"}, foo_callbacks, resource_decoder);
+ NiceMock<MockSubscriptionCallbacks> bar_callbacks;
+ auto bar_sub = grpc_mux_->addWatch(v2_type_url, {"y", "z"}, bar_callbacks, resource_decoder);
+ EXPECT_CALL(*async_client_, startRaw(_, _, _, _)).WillOnce(Return(&async_stream_));
+ expectSendMessage(v2_type_url, {"y", "z", "x"}, "", true);
+ grpc_mux_->start();
+
+ {
+ // Send resource with v3 type url, should invoke onConfigUpdate.
+ auto response = std::make_unique<envoy::service::discovery::v3::DiscoveryResponse>();
+ response->set_type_url(v3_type_url);
+ response->set_version_info("1");
+ envoy::config::endpoint::v3::ClusterLoadAssignment load_assignment;
+ load_assignment.set_cluster_name("x");
+ response->add_resources()->PackFrom(load_assignment);
+ EXPECT_CALL(bar_callbacks, onConfigUpdate(_, "1")).Times(0);
+ EXPECT_CALL(foo_callbacks, onConfigUpdate(_, "1"))
+ .WillOnce(Invoke([&load_assignment](const std::vector<DecodedResourceRef>& resources,
+ const std::string&) {
+ EXPECT_EQ(1, resources.size());
+ const auto& expected_assignment =
+ dynamic_cast<const envoy::config::endpoint::v3::ClusterLoadAssignment&>(
+ resources[0].get().resource());
+ EXPECT_TRUE(TestUtility::protoEqual(expected_assignment, load_assignment));
+ }));
+ expectSendMessage(v2_type_url, {"y", "z", "x"}, "1");
+ grpc_mux_->grpcStreamForTest().onReceiveMessage(std::move(response));
+ }
+
+ {
+ auto response = std::make_unique<envoy::service::discovery::v3::DiscoveryResponse>();
+ response->set_type_url(v3_type_url);
+ response->set_version_info("2");
+ envoy::config::endpoint::v3::ClusterLoadAssignment load_assignment_x;
+ load_assignment_x.set_cluster_name("x");
+ response->add_resources()->PackFrom(load_assignment_x);
+ envoy::config::endpoint::v3::ClusterLoadAssignment load_assignment_y;
+ load_assignment_y.set_cluster_name("y");
+ response->add_resources()->PackFrom(load_assignment_y);
+ envoy::config::endpoint::v3::ClusterLoadAssignment load_assignment_z;
+ load_assignment_z.set_cluster_name("z");
+ response->add_resources()->PackFrom(load_assignment_z);
+ EXPECT_CALL(bar_callbacks, onConfigUpdate(_, "2"))
+ .WillOnce(Invoke([&load_assignment_y, &load_assignment_z](
+ const std::vector<DecodedResourceRef>& resources, const std::string&) {
+ EXPECT_EQ(2, resources.size());
+ const auto& expected_assignment =
+ dynamic_cast<const envoy::config::endpoint::v3::ClusterLoadAssignment&>(
+ resources[0].get().resource());
+ EXPECT_TRUE(TestUtility::protoEqual(expected_assignment, load_assignment_y));
+ const auto& expected_assignment_1 =
+ dynamic_cast<const envoy::config::endpoint::v3::ClusterLoadAssignment&>(
+ resources[1].get().resource());
+ EXPECT_TRUE(TestUtility::protoEqual(expected_assignment_1, load_assignment_z));
+ }));
+ EXPECT_CALL(foo_callbacks, onConfigUpdate(_, "2"))
+ .WillOnce(Invoke([&load_assignment_x, &load_assignment_y](
+ const std::vector<DecodedResourceRef>& resources, const std::string&) {
+ EXPECT_EQ(2, resources.size());
+ const auto& expected_assignment =
+ dynamic_cast<const envoy::config::endpoint::v3::ClusterLoadAssignment&>(
+ resources[0].get().resource());
+ EXPECT_TRUE(TestUtility::protoEqual(expected_assignment, load_assignment_x));
+ const auto& expected_assignment_1 =
+ dynamic_cast<const envoy::config::endpoint::v3::ClusterLoadAssignment&>(
+ resources[1].get().resource());
+ EXPECT_TRUE(TestUtility::protoEqual(expected_assignment_1, load_assignment_y));
+ }));
+ expectSendMessage(v2_type_url, {"y", "z", "x"}, "2");
+ grpc_mux_->grpcStreamForTest().onReceiveMessage(std::move(response));
+ }
+
+ expectSendMessage(v2_type_url, {"x", "y"}, "2");
+ expectSendMessage(v2_type_url, {}, "2");
+}
+
 } // namespace
 } // namespace Config
 } // namespace Envoy

test/common/config/new_grpc_mux_impl_test.cc

@@ -162,17 +162,40 @@ TEST_F(NewGrpcMuxImplTest, ConfigUpdateWithNotFoundResponse) {
  response->set_type_url(type_url);
  response->set_system_version_info("1");
 
- response->add_resources();
- response->mutable_resources()->at(0).set_name("not-found");
- response->mutable_resources()->at(0).add_aliases("domain1.test");
+ envoy::config::route::v3::VirtualHost vhost;
+ vhost.set_name("vhost_1");
 
- grpc_mux_->onDiscoveryResponse(std::move(response), control_plane_stats_);
+ response->add_resources()->mutable_resource()->PackFrom(vhost);
+}
 
- const auto& subscriptions = grpc_mux_->subscriptions();
- auto sub = subscriptions.find(type_url);
+// Watch v2 resource type_url, receive discovery response with v3 resource type_url.
+TEST_F(NewGrpcMuxImplTest, V3ResourceResponseV2ResourceWatch) {
+ setup();
 
- EXPECT_TRUE(sub != subscriptions.end());
- watch->update({});
+ // Watch for v2 resource type_url.
+ const std::string& type_url = Config::TypeUrl::get().ClusterLoadAssignment;
+ auto watch = grpc_mux_->addWatch(type_url, {}, callbacks_, resource_decoder_);
+
+ EXPECT_CALL(*async_client_, startRaw(_, _, _, _)).WillOnce(Return(&async_stream_));
+ grpc_mux_->start();
+ {
+ auto response = std::make_unique<envoy::service::discovery::v3::DeltaDiscoveryResponse>();
+ response->set_system_version_info("1");
+ envoy::config::endpoint::v3::ClusterLoadAssignment load_assignment;
+ load_assignment.set_cluster_name("x");
+ response->add_resources()->mutable_resource()->PackFrom(load_assignment);
+ // Send response that contains resource with v3 type url.
+ response->set_type_url("type.googleapis.com/envoy.config.endpoint.v3.ClusterLoadAssignment");
+ EXPECT_CALL(callbacks_, onConfigUpdate(_, _, "1"))
+ .WillOnce(Invoke([&load_assignment](const std::vector<DecodedResourceRef>& added_resources,
+ const Protobuf::RepeatedPtrField<std::string>&,
+ const std::string&) {
+ EXPECT_EQ(1, added_resources.size());
+ EXPECT_TRUE(
+ TestUtility::protoEqual(added_resources[0].get().resource(), load_assignment));
+ }));
+ grpc_mux_->onDiscoveryResponse(std::move(response), control_plane_stats_);
+ }
 }
 
 } // namespace

test/integration/ads_integration_test.cc

@@ -115,6 +115,31 @@ TEST_P(AdsIntegrationTest, Failure) {
  makeSingleRequest();
 }
 
+// Validate that xds can support a mix of v2 and v3 type url.
+TEST_P(AdsIntegrationTest, MixV2V3TypeUrlInDiscoveryResponse) {
+ initialize();
+ // Discovery response with v3 type url.
+ sendDiscoveryResponse<envoy::config::cluster::v3::Cluster>(
+ "type.googleapis.com/envoy.config.cluster.v3.Cluster", {buildCluster("cluster_0")},
+ {buildCluster("cluster_0")}, {}, "1", false);
+ // Discovery response with v2 type url.
+ sendDiscoveryResponse<envoy::config::endpoint::v3::ClusterLoadAssignment>(
+ Config::TypeUrl::get().ClusterLoadAssignment, {buildClusterLoadAssignment("cluster_0")},
+ {buildClusterLoadAssignment("cluster_0")}, {}, "1");
+ // Discovery response with v3 type url.
+ sendDiscoveryResponse<envoy::config::listener::v3::Listener>(
+ "type.googleapis.com/envoy.config.listener.v3.Listener",
+ {buildListener("listener_0", "route_config_0")},
+ {buildListener("listener_0", "route_config_0")}, {}, "1", false);
+ // Discovery response with v2 type url.
+ sendDiscoveryResponse<envoy::config::route::v3::RouteConfiguration>(
+ Config::TypeUrl::get().RouteConfiguration, {buildRouteConfig("route_config_0", "cluster_0")},
+ {buildRouteConfig("route_config_0", "cluster_0")}, {}, "1");
+ test_server_->waitForCounterGe("listener_manager.listener_create_success", 1);
+ // Validate that we can process a request.
+ makeSingleRequest();
+}
+
 // Validate that the request with duplicate listeners is rejected.
 TEST_P(AdsIntegrationTest, DuplicateWarmingListeners) {
  initialize();