diff --git a/include/envoy/stream_info/stream_info.h b/include/envoy/stream_info/stream_info.h index 1558503ccfe5..442d8a1699a1 100644 --- a/include/envoy/stream_info/stream_info.h +++ b/include/envoy/stream_info/stream_info.h @@ -109,6 +109,8 @@ struct ResponseCodeDetailValues { // The request was rejected because it attempted an unsupported upgrade. const std::string UpgradeFailed = "upgrade_failed"; + // The request was rejected by the HCM because there was no route configuration found. + const std::string RouteConfigurationNotFound = "route_configuration_not_found"; // The request was rejected by the router filter because there was no route found. const std::string RouteNotFound = "route_not_found"; // A direct response was generated by the router filter. diff --git a/source/common/http/BUILD b/source/common/http/BUILD index 1dcf5045fa26..65a8d6da1473 100644 --- a/source/common/http/BUILD +++ b/source/common/http/BUILD @@ -164,6 +164,7 @@ envoy_cc_library( "//include/envoy/router:rds_interface", "//include/envoy/router:scopes_interface", "//include/envoy/runtime:runtime_interface", + "//include/envoy/server:admin_interface", "//include/envoy/server:overload_manager_interface", "//include/envoy/ssl:connection_interface", "//include/envoy/stats:stats_interface", diff --git a/source/common/http/conn_manager_impl.cc b/source/common/http/conn_manager_impl.cc index 8ee08cf2b988..9f3dd522a784 100644 --- a/source/common/http/conn_manager_impl.cc +++ b/source/common/http/conn_manager_impl.cc @@ -12,6 +12,7 @@ #include "envoy/event/dispatcher.h" #include "envoy/network/drain_decision.h" #include "envoy/router/router.h" +#include "envoy/server/admin.h" #include "envoy/ssl/connection.h" #include "envoy/stats/scope.h" #include "envoy/tracing/http_tracer.h" @@ -431,12 +432,25 @@ void ConnectionManagerImpl::chargeTracingStats(const Tracing::Reason& tracing_re ConnectionManagerImpl::ActiveStream::ActiveStream(ConnectionManagerImpl& connection_manager) : connection_manager_(connection_manager), - snapped_route_config_(connection_manager.config_.routeConfigProvider()->config()), stream_id_(connection_manager.random_generator_.random()), request_response_timespan_(new Stats::Timespan( connection_manager_.stats_.named_.downstream_rq_time_, connection_manager_.timeSource())), stream_info_(connection_manager_.codec_->protocol(), connection_manager_.timeSource()), upstream_options_(std::make_shared()) { + // For Admin thread, we don't use routeConfigProvider or SRDS route provider. + ASSERT(dynamic_cast(&connection_manager_.config_) != nullptr || + ((connection_manager.config_.routeConfigProvider() == nullptr && + connection_manager.config_.scopedRouteConfigProvider() != nullptr) || + (connection_manager.config_.routeConfigProvider() != nullptr && + connection_manager.config_.scopedRouteConfigProvider() == nullptr)), + "Either routeConfigProvider or scopedRouteConfigProvider should be set in " + "ConnectionManagerImpl."); + if (connection_manager.config_.routeConfigProvider() != nullptr) { + snapped_route_config_ = connection_manager.config_.routeConfigProvider()->config(); + } else if (connection_manager.config_.scopedRouteConfigProvider() != nullptr) { + snapped_scoped_routes_config_ = + connection_manager_.config_.scopedRouteConfigProvider()->config(); + } ScopeTrackerScopeState scope(this, connection_manager_.read_callbacks_->connection().dispatcher()); @@ -612,6 +626,35 @@ void ConnectionManagerImpl::ActiveStream::decodeHeaders(HeaderMapPtr&& headers, ScopeTrackerScopeState scope(this, connection_manager_.read_callbacks_->connection().dispatcher()); request_headers_ = std::move(headers); + // For Admin thread, we don't use routeConfigProvider or SRDS route provider. + if (dynamic_cast(&connection_manager_.config_) == nullptr && + connection_manager_.config_.scopedRouteConfigProvider() != nullptr) { + ASSERT(snapped_route_config_ == nullptr, + "Route config already latched to the active stream when scoped RDS is enabled."); + if (snapped_scoped_routes_config_ == nullptr) { + ENVOY_STREAM_LOG(trace, "snapped scoped routes config is null when SRDS is enabled.", *this); + // Stop decoding now. + maybeEndDecode(true); + sendLocalReply( + Grpc::Common::hasGrpcContentType(*request_headers_), Http::Code::InternalServerError, + "scoped routes config not set when SRDS is enabled", nullptr, is_head_request_, + absl::nullopt, StreamInfo::ResponseCodeDetails::get().RouteConfigurationNotFound); + return; + } + snapped_route_config_ = snapped_scoped_routes_config_->getRouteConfig(*request_headers_); + // NOTE: if a RDS subscription hasn't got a RouteConfiguration back, a Router::NullConfigImpl is + // returned, in that case we let it pass. + if (snapped_route_config_ == nullptr) { + ENVOY_STREAM_LOG(trace, "can't find SRDS scope.", *this); + // Stop decoding now. + maybeEndDecode(true); + sendLocalReply(Grpc::Common::hasGrpcContentType(*request_headers_), Http::Code::NotFound, + "route scope not found", nullptr, is_head_request_, absl::nullopt, + StreamInfo::ResponseCodeDetails::get().RouteConfigurationNotFound); + return; + } + } + if (Http::Headers::get().MethodValues.Head == request_headers_->Method()->value().getStringView()) { is_head_request_ = true; diff --git a/source/common/http/conn_manager_impl.h b/source/common/http/conn_manager_impl.h index 84dde922406e..5691f05846a3 100644 --- a/source/common/http/conn_manager_impl.h +++ b/source/common/http/conn_manager_impl.h @@ -585,7 +585,7 @@ class ConnectionManagerImpl : Logger::Loggable, ConnectionManagerImpl& connection_manager_; Router::ConfigConstSharedPtr snapped_route_config_; - Router::ScopedConfigConstSharedPtr snapped_scoped_route_config_; + Router::ScopedConfigConstSharedPtr snapped_scoped_routes_config_; Tracing::SpanPtr active_span_; const uint64_t stream_id_; StreamEncoder* response_encoder_{}; diff --git a/source/common/router/scoped_config_impl.h b/source/common/router/scoped_config_impl.h index f6197f09abdc..39cdb44b1361 100644 --- a/source/common/router/scoped_config_impl.h +++ b/source/common/router/scoped_config_impl.h @@ -158,7 +158,7 @@ class ScopedRouteInfo { const std::string& scopeName() const { return config_proto_.name(); } private: - envoy::api::v2::ScopedRouteConfiguration config_proto_; + const envoy::api::v2::ScopedRouteConfiguration config_proto_; ScopeKey scope_key_; ConfigConstSharedPtr route_config_; }; diff --git a/source/common/router/scoped_rds.cc b/source/common/router/scoped_rds.cc index 9e3c8079c1a1..daeb490ccf01 100644 --- a/source/common/router/scoped_rds.cc +++ b/source/common/router/scoped_rds.cc @@ -317,7 +317,6 @@ void ScopedRdsConfigSubscription::onConfigUpdate( } onConfigUpdate(to_add_repeated, to_remove_repeated, version_info); } // namespace Router - ScopedRdsConfigProvider::ScopedRdsConfigProvider( ScopedRdsConfigSubscriptionSharedPtr&& subscription) : MutableConfigProviderCommonBase(std::move(subscription), ConfigProvider::ApiType::Delta) {} diff --git a/test/common/http/BUILD b/test/common/http/BUILD index 56eefc2b25a6..a93962a212c8 100644 --- a/test/common/http/BUILD +++ b/test/common/http/BUILD @@ -129,17 +129,6 @@ envoy_cc_test_library( ], ) -envoy_cc_test_library( - name = "conn_manager_impl_common_lib", - hdrs = ["conn_manager_impl_common.h"], - deps = [ - "//include/envoy/common:time_interface", - "//include/envoy/config:config_provider_interface", - "//include/envoy/router:rds_interface", - "//test/mocks/router:router_mocks", - ], -) - envoy_proto_library( name = "conn_manager_impl_fuzz_proto", srcs = ["conn_manager_impl_fuzz.proto"], @@ -153,7 +142,6 @@ envoy_cc_fuzz_test( srcs = ["conn_manager_impl_fuzz_test.cc"], corpus = "conn_manager_impl_corpus", deps = [ - ":conn_manager_impl_common_lib", ":conn_manager_impl_fuzz_proto_cc", "//source/common/common:empty_string", "//source/common/http:conn_manager_lib", @@ -166,6 +154,7 @@ envoy_cc_fuzz_test( "//test/mocks/http:http_mocks", "//test/mocks/local_info:local_info_mocks", "//test/mocks/network:network_mocks", + "//test/mocks/router:router_mocks", "//test/mocks/runtime:runtime_mocks", "//test/mocks/ssl:ssl_mocks", "//test/mocks/tracing:tracing_mocks", @@ -179,7 +168,6 @@ envoy_cc_test( name = "conn_manager_impl_test", srcs = ["conn_manager_impl_test.cc"], deps = [ - ":conn_manager_impl_common_lib", "//include/envoy/access_log:access_log_interface", "//include/envoy/buffer:buffer_interface", "//include/envoy/event:dispatcher_interface", @@ -206,6 +194,7 @@ envoy_cc_test( "//test/mocks/http:http_mocks", "//test/mocks/local_info:local_info_mocks", "//test/mocks/network:network_mocks", + "//test/mocks/router:router_mocks", "//test/mocks/runtime:runtime_mocks", "//test/mocks/server:server_mocks", "//test/mocks/ssl:ssl_mocks", diff --git a/test/common/http/conn_manager_impl_common.h b/test/common/http/conn_manager_impl_common.h deleted file mode 100644 index 1f68cc59412d..000000000000 --- a/test/common/http/conn_manager_impl_common.h +++ /dev/null @@ -1,54 +0,0 @@ -#pragma once - -#include - -#include "envoy/common/time.h" -#include "envoy/config/config_provider.h" -#include "envoy/router/rds.h" - -#include "test/mocks/router/mocks.h" - -#include "gmock/gmock.h" - -using testing::NiceMock; - -namespace Envoy { -namespace Http { -namespace ConnectionManagerImplHelper { - -// Test RouteConfigProvider that returns a mocked config. -struct RouteConfigProvider : public Router::RouteConfigProvider { - RouteConfigProvider(TimeSource& time_source) : time_source_(time_source) {} - - // Router::RouteConfigProvider - Router::ConfigConstSharedPtr config() override { return route_config_; } - absl::optional configInfo() const override { return {}; } - SystemTime lastUpdated() const override { return time_source_.systemTime(); } - void onConfigUpdate() override {} - - TimeSource& time_source_; - std::shared_ptr route_config_{new NiceMock()}; -}; - -// Test ScopedRouteConfigProvider that returns a mocked config. -struct ScopedRouteConfigProvider : public Config::ConfigProvider { - ScopedRouteConfigProvider(TimeSource& time_source) - : config_(std::make_shared()), time_source_(time_source) {} - - ~ScopedRouteConfigProvider() override = default; - - // Config::ConfigProvider - SystemTime lastUpdated() const override { return time_source_.systemTime(); } - const Protobuf::Message* getConfigProto() const override { return nullptr; } - Envoy::Config::ConfigProvider::ConfigProtoVector getConfigProtos() const override { return {}; } - std::string getConfigVersion() const override { return ""; } - ConfigConstSharedPtr getConfig() const override { return config_; } - ApiType apiType() const override { return ApiType::Delta; } - - std::shared_ptr config_; - TimeSource& time_source_; -}; - -} // namespace ConnectionManagerImplHelper -} // namespace Http -} // namespace Envoy diff --git a/test/common/http/conn_manager_impl_fuzz_test.cc b/test/common/http/conn_manager_impl_fuzz_test.cc index 5c2bf01fe1ab..f7e3860e3da6 100644 --- a/test/common/http/conn_manager_impl_fuzz_test.cc +++ b/test/common/http/conn_manager_impl_fuzz_test.cc @@ -20,7 +20,6 @@ #include "common/network/address_impl.h" #include "common/network/utility.h" -#include "test/common/http/conn_manager_impl_common.h" #include "test/common/http/conn_manager_impl_fuzz.pb.h" #include "test/fuzz/fuzz_runner.h" #include "test/fuzz/utility.h" @@ -29,6 +28,7 @@ #include "test/mocks/http/mocks.h" #include "test/mocks/local_info/mocks.h" #include "test/mocks/network/mocks.h" +#include "test/mocks/router/mocks.h" #include "test/mocks/runtime/mocks.h" #include "test/mocks/ssl/mocks.h" #include "test/mocks/tracing/mocks.h" @@ -46,13 +46,15 @@ namespace Http { class FuzzConfig : public ConnectionManagerConfig { public: FuzzConfig() - : route_config_provider_(time_system_), scoped_route_config_provider_(time_system_), - stats_{{ALL_HTTP_CONN_MAN_STATS(POOL_COUNTER(fake_stats_), POOL_GAUGE(fake_stats_), + : stats_{{ALL_HTTP_CONN_MAN_STATS(POOL_COUNTER(fake_stats_), POOL_GAUGE(fake_stats_), POOL_HISTOGRAM(fake_stats_))}, "", fake_stats_}, tracing_stats_{CONN_MAN_TRACING_STATS(POOL_COUNTER(fake_stats_))}, listener_stats_{CONN_MAN_LISTENER_STATS(POOL_COUNTER(fake_stats_))} { + ON_CALL(route_config_provider_, lastUpdated()).WillByDefault(Return(time_system_.systemTime())); + ON_CALL(scoped_route_config_provider_, lastUpdated()) + .WillByDefault(Return(time_system_.systemTime())); access_logs_.emplace_back(std::make_shared>()); } @@ -85,9 +87,17 @@ class FuzzConfig : public ConnectionManagerConfig { std::chrono::milliseconds streamIdleTimeout() const override { return stream_idle_timeout_; } std::chrono::milliseconds requestTimeout() const override { return request_timeout_; } std::chrono::milliseconds delayedCloseTimeout() const override { return delayed_close_timeout_; } - Router::RouteConfigProvider* routeConfigProvider() override { return &route_config_provider_; } + Router::RouteConfigProvider* routeConfigProvider() override { + if (use_srds_) { + return nullptr; + } + return &route_config_provider_; + } Config::ConfigProvider* scopedRouteConfigProvider() override { - return &scoped_route_config_provider_; + if (use_srds_) { + return &scoped_route_config_provider_; + } + return nullptr; } const std::string& serverName() override { return server_name_; } ConnectionManagerStats& stats() override { return stats_; } @@ -120,8 +130,9 @@ class FuzzConfig : public ConnectionManagerConfig { NiceMock filter_factory_; Event::SimulatedTimeSystem time_system_; SlowDateProviderImpl date_provider_{time_system_}; - ConnectionManagerImplHelper::RouteConfigProvider route_config_provider_; - ConnectionManagerImplHelper::ScopedRouteConfigProvider scoped_route_config_provider_; + bool use_srds_{}; + Router::MockRouteConfigProvider route_config_provider_; + Router::MockScopedRouteConfigProvider scoped_route_config_provider_; std::string server_name_; Stats::IsolatedStoreImpl fake_stats_; ConnectionManagerStats stats_; diff --git a/test/common/http/conn_manager_impl_test.cc b/test/common/http/conn_manager_impl_test.cc index 083bbbe73753..34c69554436a 100644 --- a/test/common/http/conn_manager_impl_test.cc +++ b/test/common/http/conn_manager_impl_test.cc @@ -26,13 +26,13 @@ #include "extensions/access_loggers/file/file_access_log_impl.h" -#include "test/common/http/conn_manager_impl_common.h" #include "test/mocks/access_log/mocks.h" #include "test/mocks/buffer/mocks.h" #include "test/mocks/common.h" #include "test/mocks/http/mocks.h" #include "test/mocks/local_info/mocks.h" #include "test/mocks/network/mocks.h" +#include "test/mocks/router/mocks.h" #include "test/mocks/runtime/mocks.h" #include "test/mocks/server/mocks.h" #include "test/mocks/ssl/mocks.h" @@ -69,9 +69,7 @@ namespace Http { class HttpConnectionManagerImplTest : public testing::Test, public ConnectionManagerConfig { public: HttpConnectionManagerImplTest() - : route_config_provider_(test_time_.timeSystem()), - scoped_route_config_provider_(test_time_.timeSystem()), - http_context_(fake_stats_.symbolTable()), access_log_path_("dummy_path"), + : http_context_(fake_stats_.symbolTable()), access_log_path_("dummy_path"), access_logs_{ AccessLog::InstanceSharedPtr{new Extensions::AccessLoggers::File::FileAccessLog( access_log_path_, {}, AccessLog::AccessLogFormatUtils::defaultAccessLogFormatter(), @@ -86,6 +84,10 @@ class HttpConnectionManagerImplTest : public testing::Test, public ConnectionMan http_context_.setTracer(tracer_); + ON_CALL(route_config_provider_, lastUpdated()) + .WillByDefault(Return(test_time_.timeSystem().systemTime())); + ON_CALL(scoped_route_config_provider_, lastUpdated()) + .WillByDefault(Return(test_time_.timeSystem().systemTime())); // response_encoder_ is not a NiceMock on purpose. This prevents complaining about this // method only. EXPECT_CALL(response_encoder_, getStream()).Times(AtLeast(0)); @@ -256,9 +258,18 @@ class HttpConnectionManagerImplTest : public testing::Test, public ConnectionMan std::chrono::milliseconds streamIdleTimeout() const override { return stream_idle_timeout_; } std::chrono::milliseconds requestTimeout() const override { return request_timeout_; } std::chrono::milliseconds delayedCloseTimeout() const override { return delayed_close_timeout_; } - Router::RouteConfigProvider* routeConfigProvider() override { return &route_config_provider_; } + bool use_srds_{}; + Router::RouteConfigProvider* routeConfigProvider() override { + if (use_srds_) { + return nullptr; + } + return &route_config_provider_; + } Config::ConfigProvider* scopedRouteConfigProvider() override { - return &scoped_route_config_provider_; + if (use_srds_) { + return &scoped_route_config_provider_; + } + return nullptr; } const std::string& serverName() override { return server_name_; } ConnectionManagerStats& stats() override { return stats_; } @@ -284,8 +295,9 @@ class HttpConnectionManagerImplTest : public testing::Test, public ConnectionMan bool shouldMergeSlashes() const override { return merge_slashes_; } DangerousDeprecatedTestTime test_time_; - ConnectionManagerImplHelper::RouteConfigProvider route_config_provider_; - ConnectionManagerImplHelper::ScopedRouteConfigProvider scoped_route_config_provider_; + NiceMock route_config_provider_; + std::shared_ptr route_config_{new NiceMock()}; + NiceMock scoped_route_config_provider_; NiceMock tracer_; Stats::IsolatedStoreImpl fake_stats_; Http::ContextImpl http_context_; @@ -4413,5 +4425,169 @@ TEST_F(HttpConnectionManagerImplTest, TestSessionTrace) { } } +TEST_F(HttpConnectionManagerImplTest, TestSRDSRouteNotFound) { + setup(false, ""); + use_srds_ = true; + + EXPECT_CALL(*static_cast( + scopedRouteConfigProvider()->config().get()), + getRouteConfig(_)) + .Times(1) + .WillOnce(Return(nullptr)); + EXPECT_CALL(*codec_, dispatch(_)).Times(1).WillOnce(Invoke([&](Buffer::Instance& data) -> void { + StreamDecoder* decoder = &conn_manager_->newStream(response_encoder_); + HeaderMapPtr headers{ + new TestHeaderMapImpl{{":authority", "host"}, {":method", "GET"}, {":path", "/foo"}}}; + decoder->decodeHeaders(std::move(headers), true); + data.drain(4); + })); + + EXPECT_CALL(response_encoder_, encodeHeaders(_, false)) + .WillOnce(Invoke([](const HeaderMap& headers, bool) -> void { + EXPECT_EQ("404", headers.Status()->value().getStringView()); + })); + + std::string response_body; + EXPECT_CALL(response_encoder_, encodeData(_, true)).WillOnce(AddBufferToString(&response_body)); + + Buffer::OwnedImpl fake_input("1234"); + conn_manager_->onData(fake_input, false); + EXPECT_EQ(response_body, "route scope not found"); +} + +TEST_F(HttpConnectionManagerImplTest, TestSRDSUpdate) { + setup(false, ""); + use_srds_ = true; + + EXPECT_CALL(*static_cast( + scopedRouteConfigProvider()->config().get()), + getRouteConfig(_)) + .Times(2) + .WillOnce(Return(nullptr)) + .WillOnce(Return(route_config_)); + EXPECT_CALL(*codec_, dispatch(_)) + .Times(2) // Once for no scoped routes, once for scoped routing + .WillRepeatedly(Invoke([&](Buffer::Instance& data) -> void { + StreamDecoder* decoder = &conn_manager_->newStream(response_encoder_); + HeaderMapPtr headers{ + new TestHeaderMapImpl{{":authority", "host"}, {":method", "GET"}, {":path", "/foo"}}}; + decoder->decodeHeaders(std::move(headers), true); + data.drain(4); + })); + EXPECT_CALL(response_encoder_, encodeHeaders(_, false)) + .WillOnce(Invoke([](const HeaderMap& headers, bool) -> void { + EXPECT_EQ("404", headers.Status()->value().getStringView()); + })); + + std::string response_body; + EXPECT_CALL(response_encoder_, encodeData(_, true)).WillOnce(AddBufferToString(&response_body)); + + Buffer::OwnedImpl fake_input("1234"); + conn_manager_->onData(fake_input, false); + EXPECT_EQ(response_body, "route scope not found"); + + // Now route config provider returns something. + setupFilterChain(1, 0); // Recreate the chain for second stream. + const std::string fake_cluster1_name = "fake_cluster1"; + std::shared_ptr route1 = std::make_shared>(); + EXPECT_CALL(route1->route_entry_, clusterName()).WillRepeatedly(ReturnRef(fake_cluster1_name)); + std::shared_ptr fake_cluster1 = + std::make_shared>(); + EXPECT_CALL(cluster_manager_, get(_)).WillOnce(Return(fake_cluster1.get())); + EXPECT_CALL(*route_config_, route(_, _)).Times(1).WillOnce(Return(route1)); + EXPECT_CALL(*decoder_filters_[0], decodeHeaders(_, true)) + .WillOnce(InvokeWithoutArgs([&]() -> FilterHeadersStatus { + EXPECT_EQ(route1, decoder_filters_[0]->callbacks_->route()); + EXPECT_EQ(route1->routeEntry(), decoder_filters_[0]->callbacks_->streamInfo().routeEntry()); + EXPECT_EQ(fake_cluster1->info(), decoder_filters_[0]->callbacks_->clusterInfo()); + return FilterHeadersStatus::StopIteration; + })); + EXPECT_CALL(*decoder_filters_[0], decodeComplete()); + Buffer::OwnedImpl fake_input2("1234"); + conn_manager_->onData(fake_input2, false); +} + +TEST_F(HttpConnectionManagerImplTest, TestSRDSRouteFound) { + setup(false, ""); + setupFilterChain(1, 0); + use_srds_ = true; + + const std::string fake_cluster1_name = "fake_cluster1"; + std::shared_ptr route1 = std::make_shared>(); + EXPECT_CALL(route1->route_entry_, clusterName()).WillRepeatedly(ReturnRef(fake_cluster1_name)); + std::shared_ptr fake_cluster1 = + std::make_shared>(); + EXPECT_CALL(cluster_manager_, get(_)).WillOnce(Return(fake_cluster1.get())); + EXPECT_CALL(*scopedRouteConfigProvider()->config(), getRouteConfig(_)) + .Times(1); + EXPECT_CALL( + *static_cast( + scopedRouteConfigProvider()->config()->route_config_.get()), + route(_, _)) + .Times(1) + .WillOnce(Return(route1)); + StreamDecoder* decoder = nullptr; + EXPECT_CALL(*codec_, dispatch(_)).Times(1).WillOnce(Invoke([&](Buffer::Instance& data) -> void { + decoder = &conn_manager_->newStream(response_encoder_); + HeaderMapPtr headers{ + new TestHeaderMapImpl{{":authority", "host"}, {":method", "GET"}, {":path", "/foo"}}}; + decoder->decodeHeaders(std::move(headers), true); + data.drain(4); + })); + EXPECT_CALL(*decoder_filters_[0], decodeHeaders(_, true)) + .WillOnce(InvokeWithoutArgs([&]() -> FilterHeadersStatus { + EXPECT_EQ(route1, decoder_filters_[0]->callbacks_->route()); + EXPECT_EQ(route1->routeEntry(), decoder_filters_[0]->callbacks_->streamInfo().routeEntry()); + EXPECT_EQ(fake_cluster1->info(), decoder_filters_[0]->callbacks_->clusterInfo()); + return FilterHeadersStatus::StopIteration; + })); + EXPECT_CALL(*decoder_filters_[0], decodeComplete()); + + Buffer::OwnedImpl fake_input("1234"); + conn_manager_->onData(fake_input, false); +} + +class HttpConnectionManagerImplDeathTest : public HttpConnectionManagerImplTest { +public: + Router::RouteConfigProvider* routeConfigProvider() override { + return route_config_provider2_.get(); + } + Config::ConfigProvider* scopedRouteConfigProvider() override { + return scoped_route_config_provider2_.get(); + } + + std::shared_ptr> route_config_provider2_; + std::shared_ptr> scoped_route_config_provider2_; +}; + +TEST_F(HttpConnectionManagerImplDeathTest, InvalidConnectionManagerConfig) { + setup(false, ""); + + Buffer::OwnedImpl fake_input("1234"); + EXPECT_CALL(*codec_, dispatch(_)).WillRepeatedly(Invoke([&](Buffer::Instance&) -> void { + conn_manager_->newStream(response_encoder_); + })); + // Either RDS or SRDS should be set. + EXPECT_DEBUG_DEATH(conn_manager_->onData(fake_input, false), + "Either routeConfigProvider or scopedRouteConfigProvider should be set in " + "ConnectionManagerImpl."); + + route_config_provider2_ = std::make_shared>(); + + // Only route config provider valid. + EXPECT_NO_THROW(conn_manager_->onData(fake_input, false)); + + scoped_route_config_provider2_ = + std::make_shared>(); + // Can't have RDS and SRDS provider in the same time. + EXPECT_DEBUG_DEATH(conn_manager_->onData(fake_input, false), + "Either routeConfigProvider or scopedRouteConfigProvider should be set in " + "ConnectionManagerImpl."); + + scoped_route_config_provider2_.reset(); + // Only scoped route config provider valid. + EXPECT_NO_THROW(conn_manager_->onData(fake_input, false)); +} + } // namespace Http } // namespace Envoy diff --git a/test/common/router/scoped_rds_test.cc b/test/common/router/scoped_rds_test.cc index dfe3c4634806..c71ec76e0ad9 100644 --- a/test/common/router/scoped_rds_test.cc +++ b/test/common/router/scoped_rds_test.cc @@ -211,6 +211,10 @@ route_configuration_name: foo_routes parseScopedRouteConfigurationFromYaml(*resources.Add(), config_yaml); EXPECT_THROW(srds_subscription_->onConfigUpdate(resources, "1"), ProtoValidationException); + EXPECT_THROW_WITH_REGEX( + srds_subscription_->onConfigUpdate(anyToResource(resources, "1"), {}, "1"), EnvoyException, + "Error adding/updating scoped route\\(s\\): Proto constraint validation failed.*"); + EXPECT_THROW_WITH_REGEX( srds_subscription_->onConfigUpdate(anyToResource(resources, "1"), {}, "1"), EnvoyException, "Error adding/updating scoped route\\(s\\): Proto constraint validation failed.*"); diff --git a/test/integration/scoped_rds_integration_test.cc b/test/integration/scoped_rds_integration_test.cc index 11ff06916cc3..17d59e9b391a 100644 --- a/test/integration/scoped_rds_integration_test.cc +++ b/test/integration/scoped_rds_integration_test.cc @@ -4,6 +4,7 @@ #include "test/common/grpc/grpc_client_integration.h" #include "test/integration/http_integration.h" +#include "test/test_common/printers.h" #include "gmock/gmock.h" #include "gtest/gtest.h" @@ -17,7 +18,7 @@ class ScopedRdsIntegrationTest : public HttpIntegrationTest, struct FakeUpstreamInfo { FakeHttpConnectionPtr connection_; FakeUpstream* upstream_{}; - FakeStreamPtr stream_; + absl::flat_hash_map stream_by_resource_name_; }; ScopedRdsIntegrationTest() @@ -29,7 +30,15 @@ class ScopedRdsIntegrationTest : public HttpIntegrationTest, } void initialize() override { + // Setup two upstream hosts, one for each cluster. + setUpstreamCount(2); + config_helper_.addConfigModifier([](envoy::config::bootstrap::v2::Bootstrap& bootstrap) { + // Add the static cluster to serve SRDS. + auto* cluster_1 = bootstrap.mutable_static_resources()->add_clusters(); + cluster_1->MergeFrom(bootstrap.static_resources().clusters()[0]); + cluster_1->set_name("cluster_1"); + // Add the static cluster to serve SRDS. auto* scoped_rds_cluster = bootstrap.mutable_static_resources()->add_clusters(); scoped_rds_cluster->MergeFrom(bootstrap.static_resources().clusters()[0]); @@ -50,9 +59,10 @@ class ScopedRdsIntegrationTest : public HttpIntegrationTest, fragments: - header_value_extractor: name: Addr + element_separator: ; element: key: x-foo-key - separator: ; + separator: = )EOF"; envoy::config::filter::network::http_connection_manager::v2::ScopedRoutes::ScopeKeyBuilder scope_key_builder; @@ -80,6 +90,40 @@ class ScopedRdsIntegrationTest : public HttpIntegrationTest, HttpIntegrationTest::initialize(); } + // Helper that verifies if given headers are in the response header map. + void verifyResponse(IntegrationStreamDecoderPtr response, const std::string& response_code, + const Http::TestHeaderMapImpl& expected_headers, + const std::string& expected_body) { + EXPECT_TRUE(response->complete()); + EXPECT_EQ(response_code, response->headers().Status()->value().getStringView()); + expected_headers.iterate( + [](const Http::HeaderEntry& header, void* context) -> Http::HeaderMap::Iterate { + auto response_headers = static_cast(context); + const Http::HeaderEntry* entry = response_headers->get( + Http::LowerCaseString{std::string(header.key().getStringView())}); + EXPECT_NE(entry, nullptr); + EXPECT_EQ(header.value().getStringView(), entry->value().getStringView()); + return Http::HeaderMap::Iterate::Continue; + }, + const_cast(static_cast(&response->headers()))); + EXPECT_EQ(response->body(), expected_body); + } + + // Helper that sends a request to Envoy, and verifies if Envoy response headers and body size is + // the same as the expected headers map. + void sendRequestAndVerifyResponse(const Http::TestHeaderMapImpl& request_headers, + const int request_size, + const Http::TestHeaderMapImpl& response_headers, + const int response_size, const int backend_idx) { + codec_client_ = makeHttpConnection(lookupPort("http")); + auto response = sendRequestAndWaitForResponse(request_headers, request_size, response_headers, + response_size, backend_idx); + verifyResponse(std::move(response), "200", response_headers, std::string(response_size, 'a')); + EXPECT_TRUE(upstream_request_->complete()); + EXPECT_EQ(request_size, upstream_request_->bodyLength()); + cleanupUpstreamAndDownstream(); + } + void createUpstreams() override { HttpIntegrationTest::createUpstreams(); // Create the SRDS upstream. @@ -108,24 +152,47 @@ class ScopedRdsIntegrationTest : public HttpIntegrationTest, resetFakeUpstreamInfo(&scoped_rds_upstream_info_); } - FakeUpstream& getRdsFakeUpstream() const { return *fake_upstreams_[2]; } + FakeUpstream& getRdsFakeUpstream() const { return *fake_upstreams_[3]; } - FakeUpstream& getScopedRdsFakeUpstream() const { return *fake_upstreams_[1]; } + FakeUpstream& getScopedRdsFakeUpstream() const { return *fake_upstreams_[2]; } - void createStream(FakeUpstreamInfo* upstream_info, FakeUpstream& upstream) { - upstream_info->upstream_ = &upstream; - AssertionResult result = - upstream_info->upstream_->waitForHttpConnection(*dispatcher_, upstream_info->connection_); - RELEASE_ASSERT(result, result.message()); - result = upstream_info->connection_->waitForNewStream(*dispatcher_, upstream_info->stream_); + void createStream(FakeUpstreamInfo* upstream_info, FakeUpstream& upstream, + const std::string& resource_name) { + if (upstream_info->upstream_ == nullptr) { + // bind upstream if not yet. + upstream_info->upstream_ = &upstream; + AssertionResult result = + upstream_info->upstream_->waitForHttpConnection(*dispatcher_, upstream_info->connection_); + RELEASE_ASSERT(result, result.message()); + } + if (!upstream_info->stream_by_resource_name_.try_emplace(resource_name, nullptr).second) { + RELEASE_ASSERT(false, + fmt::format("stream with resource name '{}' already exists!", resource_name)); + } + auto result = upstream_info->connection_->waitForNewStream( + *dispatcher_, upstream_info->stream_by_resource_name_[resource_name]); RELEASE_ASSERT(result, result.message()); - upstream_info->stream_->startGrpcStream(); + upstream_info->stream_by_resource_name_[resource_name]->startGrpcStream(); } - void createRdsStream() { createStream(&rds_upstream_info_, getRdsFakeUpstream()); } + void createRdsStream(const std::string& resource_name) { + createStream(&rds_upstream_info_, getRdsFakeUpstream(), resource_name); + } void createScopedRdsStream() { - createStream(&scoped_rds_upstream_info_, getScopedRdsFakeUpstream()); + createStream(&scoped_rds_upstream_info_, getScopedRdsFakeUpstream(), "foo-scoped-routes"); + } + + void sendRdsResponse(const std::string& route_config, const std::string& version) { + envoy::api::v2::DiscoveryResponse response; + response.set_version_info(version); + response.set_type_url(Config::TypeUrl::get().RouteConfiguration); + auto route_configuration = + TestUtility::parseYaml(route_config); + response.add_resources()->PackFrom(route_configuration); + ASSERT(rds_upstream_info_.stream_by_resource_name_[route_configuration.name()] != nullptr); + rds_upstream_info_.stream_by_resource_name_[route_configuration.name()]->sendGrpcMessage( + response); } void sendRdsResponse(const std::string& route_config, const std::string& version) { @@ -139,7 +206,7 @@ class ScopedRdsIntegrationTest : public HttpIntegrationTest, void sendScopedRdsResponse(const std::vector& resource_protos, const std::string& version) { - ASSERT(scoped_rds_upstream_info_.stream_ != nullptr); + ASSERT(scoped_rds_upstream_info_.stream_by_resource_name_["foo-scoped-routes"] != nullptr); envoy::api::v2::DiscoveryResponse response; response.set_version_info(version); @@ -150,8 +217,8 @@ class ScopedRdsIntegrationTest : public HttpIntegrationTest, TestUtility::loadFromYaml(resource_proto, scoped_route_proto); response.add_resources()->PackFrom(scoped_route_proto); } - - scoped_rds_upstream_info_.stream_->sendGrpcMessage(response); + scoped_rds_upstream_info_.stream_by_resource_name_["foo-scoped-routes"]->sendGrpcMessage( + response); } FakeUpstreamInfo scoped_rds_upstream_info_; @@ -163,20 +230,15 @@ INSTANTIATE_TEST_SUITE_P(IpVersionsAndGrpcTypes, ScopedRdsIntegrationTest, // Test that a SRDS DiscoveryResponse is successfully processed. TEST_P(ScopedRdsIntegrationTest, BasicSuccess) { - const std::string scope_route1 = R"EOF( -name: foo_scope1 -route_configuration_name: foo_route1 -key: - fragments: - - string_key: x-foo-key -)EOF"; - const std::string scope_route2 = R"EOF( -name: foo_scope2 -route_configuration_name: foo_route1 + const std::string scope_tmpl = R"EOF( +name: {} +route_configuration_name: {} key: fragments: - - string_key: x-bar-key + - string_key: {} )EOF"; + const std::string scope_route1 = fmt::format(scope_tmpl, "foo_scope1", "foo_route1", "foo-route"); + const std::string scope_route2 = fmt::format(scope_tmpl, "foo_scope2", "foo_route1", "bar-route"); const std::string route_config_tmpl = R"EOF( name: {} @@ -191,36 +253,123 @@ route_configuration_name: foo_route1 on_server_init_function_ = [&]() { createScopedRdsStream(); sendScopedRdsResponse({scope_route1, scope_route2}, "1"); - createRdsStream(); - sendRdsResponse(fmt::format(route_config_tmpl, "foo_route1", "cluster_foo_1"), "1"); - sendRdsResponse(fmt::format(route_config_tmpl, "foo_route1", "cluster_foo_2"), "2"); + createRdsStream("foo_route1"); + // CreateRdsStream waits for connection which is fired by RDS subscription. + sendRdsResponse(fmt::format(route_config_tmpl, "foo_route1", "cluster_0"), "1"); }; initialize(); + registerTestServerPorts({"http"}); + + // No scope key matches "xyz-route". + codec_client_ = makeHttpConnection(lookupPort("http")); + auto response = codec_client_->makeHeaderOnlyRequest( + Http::TestHeaderMapImpl{{":method", "GET"}, + {":path", "/meh"}, + {":authority", "host"}, + {":scheme", "http"}, + {"Addr", "x-foo-key=xyz-route"}}); + response->waitForEndStream(); + verifyResponse(std::move(response), "404", Http::TestHeaderMapImpl{}, "route scope not found"); + cleanupUpstreamAndDownstream(); + + // Test "foo-route" and 'bar-route' both gets routed to cluster_0. + test_server_->waitForCounterGe("http.config_test.rds.foo_route1.update_success", 1); + for (const std::string& scope_key : std::vector{"foo-route", "bar-route"}) { + sendRequestAndVerifyResponse( + Http::TestHeaderMapImpl{{":method", "GET"}, + {":path", "/meh"}, + {":authority", "host"}, + {":scheme", "http"}, + {"Addr", fmt::format("x-foo-key={}", scope_key)}}, + 456, Http::TestHeaderMapImpl{{":status", "200"}, {"service", scope_key}}, 123, + /*cluster_0*/ 0); + } test_server_->waitForCounterGe("http.config_test.scoped_rds.foo-scoped-routes.update_attempt", 2); test_server_->waitForCounterGe("http.config_test.scoped_rds.foo-scoped-routes.update_success", 1); // The version gauge should be set to xxHash64("1"). test_server_->waitForGaugeEq("http.config_test.scoped_rds.foo-scoped-routes.version", 13237225503670494420UL); - const std::string scope_route3 = R"EOF( -name: foo_scope3 -route_configuration_name: foo_route1 -key: - fragments: - - string_key: x-baz-key -)EOF"; - sendScopedRdsResponse({scope_route3}, "2"); + // Add a new scope scope_route3 with a brand new RouteConfiguration foo_route2. + const std::string scope_route3 = fmt::format(scope_tmpl, "foo_scope3", "foo_route2", "baz-route"); + + sendScopedRdsResponse({scope_route3, scope_route1, scope_route2}, "2"); + test_server_->waitForCounterGe("http.config_test.rds.foo_route1.update_attempt", 2); + sendRdsResponse(fmt::format(route_config_tmpl, "foo_route1", "cluster_1"), "3"); + createRdsStream("foo_route2"); + test_server_->waitForCounterGe("http.config_test.rds.foo_route2.update_attempt", 1); + sendRdsResponse(fmt::format(route_config_tmpl, "foo_route2", "cluster_0"), "1"); + test_server_->waitForCounterGe("http.config_test.rds.foo_route1.update_success", 2); + test_server_->waitForCounterGe("http.config_test.rds.foo_route2.update_success", 1); test_server_->waitForCounterGe("http.config_test.scoped_rds.foo-scoped-routes.update_success", 2); + // The version gauge should be set to xxHash64("2"). test_server_->waitForGaugeEq("http.config_test.scoped_rds.foo-scoped-routes.version", 6927017134761466251UL); - test_server_->waitForCounterGe("http.config_test.rds.foo_route1.update_attempt", 3); - sendRdsResponse(fmt::format(route_config_tmpl, "foo_route1", "cluster_foo_3"), "3"); - test_server_->waitForCounterGe("http.config_test.rds.foo_route1.update_success", 3); - // RDS updates won't affect SRDS. - test_server_->waitForGaugeEq("http.config_test.scoped_rds.foo-scoped-routes.version", - 6927017134761466251UL); - // TODO(AndresGuedez): test actual scoped routing logic; only the config handling is implemented - // at this point. + // After RDS update, requests within scope 'foo_scope1' or 'foo_scope2' get routed to 'cluster_1'. + for (const std::string& scope_key : std::vector{"foo-route", "bar-route"}) { + sendRequestAndVerifyResponse( + Http::TestHeaderMapImpl{{":method", "GET"}, + {":path", "/meh"}, + {":authority", "host"}, + {":scheme", "http"}, + {"Addr", fmt::format("x-foo-key={}", scope_key)}}, + 456, Http::TestHeaderMapImpl{{":status", "200"}, {"service", scope_key}}, 123, + /*cluster_1*/ 1); + } + // Now requests within scope 'foo_scope3' get routed to 'cluster_0'. + test_server_->waitForCounterGe("http.config_test.rds.foo_route2.update_success", 1); + sendRequestAndVerifyResponse( + Http::TestHeaderMapImpl{{":method", "GET"}, + {":path", "/meh"}, + {":authority", "host"}, + {":scheme", "http"}, + {"Addr", fmt::format("x-foo-key={}", "baz-route")}}, + 456, Http::TestHeaderMapImpl{{":status", "200"}, {"service", "bluh"}}, 123, /*cluster_0*/ 0); + + // Delete foo_scope1 and requests within the scope gets 400s. + sendScopedRdsResponse({scope_route3, scope_route2}, "3"); + test_server_->waitForCounterGe("http.config_test.scoped_rds.foo-scoped-routes.update_success", 3); + codec_client_ = makeHttpConnection(lookupPort("http")); + response = codec_client_->makeHeaderOnlyRequest( + Http::TestHeaderMapImpl{{":method", "GET"}, + {":path", "/meh"}, + {":authority", "host"}, + {":scheme", "http"}, + {"Addr", "x-foo-key=foo-route"}}); + response->waitForEndStream(); + verifyResponse(std::move(response), "404", Http::TestHeaderMapImpl{}, "route scope not found"); + cleanupUpstreamAndDownstream(); + // Add a new scope foo_scope4. + const std::string& scope_route4 = + fmt::format(scope_tmpl, "foo_scope4", "foo_route4", "xyz-route"); + sendScopedRdsResponse({scope_route3, scope_route2, scope_route4}, "4"); + test_server_->waitForCounterGe("http.config_test.scoped_rds.foo-scoped-routes.update_success", 4); + codec_client_ = makeHttpConnection(lookupPort("http")); + response = codec_client_->makeHeaderOnlyRequest( + Http::TestHeaderMapImpl{{":method", "GET"}, + {":path", "/meh"}, + {":authority", "host"}, + {":scheme", "http"}, + {"Addr", "x-foo-key=xyz-route"}}); + response->waitForEndStream(); + // Get 404 because RDS hasn't pushed route configuration "foo_route4" yet. + // But scope is found and the Router::NullConfigImpl is returned. + verifyResponse(std::move(response), "404", Http::TestHeaderMapImpl{}, ""); + cleanupUpstreamAndDownstream(); + + // RDS updated foo_route4, requests with socpe key "xyz-route" now hit cluster_1. + test_server_->waitForCounterGe("http.config_test.rds.foo_route4.update_attempt", 1); + createRdsStream("foo_route4"); + sendRdsResponse(fmt::format(route_config_tmpl, "foo_route4", "cluster_1"), "3"); + test_server_->waitForCounterGe("http.config_test.rds.foo_route4.update_success", 1); + sendRequestAndVerifyResponse( + Http::TestHeaderMapImpl{{":method", "GET"}, + {":path", "/meh"}, + {":authority", "host"}, + {":scheme", "http"}, + {"Addr", "x-foo-key=xyz-route"}}, + 456, Http::TestHeaderMapImpl{{":status", "200"}, {"service", "xyz-route"}}, 123, + /*cluster_1 */ 1); } // Test that a bad config update updates the corresponding stats. @@ -231,7 +380,7 @@ TEST_P(ScopedRdsIntegrationTest, ConfigUpdateFailure) { route_configuration_name: foo_route1 key: fragments: - - string_key: x-foo-key + - string_key: foo )EOF"; on_server_init_function_ = [this, &scope_route1]() { createScopedRdsStream(); @@ -241,6 +390,46 @@ route_configuration_name: foo_route1 test_server_->waitForCounterGe("http.config_test.scoped_rds.foo-scoped-routes.update_rejected", 1); + codec_client_ = makeHttpConnection(lookupPort("http")); + auto response = + codec_client_->makeHeaderOnlyRequest(Http::TestHeaderMapImpl{{":method", "GET"}, + {":path", "/meh"}, + {":authority", "host"}, + {":scheme", "http"}, + {"Addr", "x-foo-key=foo"}}); + response->waitForEndStream(); + verifyResponse(std::move(response), "404", Http::TestHeaderMapImpl{}, "route scope not found"); + cleanupUpstreamAndDownstream(); + + // SRDS update fixed the problem. + const std::string scope_route2 = R"EOF( +name: foo_scope1 +route_configuration_name: foo_route1 +key: + fragments: + - string_key: foo +)EOF"; + sendScopedRdsResponse({scope_route2}, "2"); + test_server_->waitForCounterGe("http.config_test.rds.foo_route1.update_attempt", 1); + createRdsStream("foo_route1"); + const std::string route_config_tmpl = R"EOF( + name: {} + virtual_hosts: + - name: integration + domains: ["*"] + routes: + - match: {{ prefix: "/" }} + route: {{ cluster: {} }} +)EOF"; + sendRdsResponse(fmt::format(route_config_tmpl, "foo_route1", "cluster_0"), "1"); + test_server_->waitForCounterGe("http.config_test.rds.foo_route1.update_success", 1); + sendRequestAndVerifyResponse( + Http::TestHeaderMapImpl{{":method", "GET"}, + {":path", "/meh"}, + {":authority", "host"}, + {":scheme", "http"}, + {"Addr", "x-foo-key=foo"}}, + 456, Http::TestHeaderMapImpl{{":status", "200"}, {"service", "bluh"}}, 123, /*cluster_0*/ 0); } } // namespace