diff --git a/changelogs/current.yaml b/changelogs/current.yaml index 9ecf0d6e48ce5..d2050924e28d4 100644 --- a/changelogs/current.yaml +++ b/changelogs/current.yaml @@ -8,6 +8,9 @@ minor_behavior_changes: bug_fixes: # *Changes expected to improve the state of the world and are unlikely to have negative effects* +- area: load_report + change: | + Fixed an issue upon load-report shutdown race with ADS stream. Introduced proper cleanup of the gRPC stream. removed_config_or_runtime: # *Normally occurs at the end of the* :ref:`deprecation period ` diff --git a/source/common/quic/envoy_quic_client_session.h b/source/common/quic/envoy_quic_client_session.h index 111d060bfd023..3db7136dd5ff8 100644 --- a/source/common/quic/envoy_quic_client_session.h +++ b/source/common/quic/envoy_quic_client_session.h @@ -105,9 +105,10 @@ class EnvoyQuicClientSession : public QuicFilterManagerConnectionImpl, quic::QuicSpdyStream* CreateIncomingStream(quic::PendingStream* pending) override; std::unique_ptr CreateQuicCryptoStream() override; bool ShouldCreateOutgoingBidirectionalStream() override { - ASSERT(quic::QuicSpdyClientSession::ShouldCreateOutgoingBidirectionalStream()); - // Prefer creating an "invalid" stream outside of current stream bounds to - // crashing when dereferencing a nullptr in QuicHttpClientConnectionImpl::newStream + // quic::QuicSpdyClientSession::ShouldCreateOutgoingBidirectionalStream() + // might return false, but we want to create the stream anyway + // because otherwise we crash dereferencing a nullptr, so we + // don't even ask it, and just return true. return true; } // QuicFilterManagerConnectionImpl diff --git a/source/common/upstream/cluster_manager_impl.h b/source/common/upstream/cluster_manager_impl.h index fd4adb4d330e5..da43a4b0be6f4 100644 --- a/source/common/upstream/cluster_manager_impl.h +++ b/source/common/upstream/cluster_manager_impl.h @@ -312,6 +312,7 @@ class ClusterManagerImpl : public ClusterManager, // Make sure we destroy all potential outgoing connections before this returns. cds_api_.reset(); xds_manager_.shutdown(); + load_stats_reporter_.reset(); active_clusters_.clear(); warming_clusters_.clear(); updateClusterCounts(); diff --git a/source/common/upstream/load_stats_reporter.cc b/source/common/upstream/load_stats_reporter.cc index 6cc62f14709ff..cce389102dba2 100644 --- a/source/common/upstream/load_stats_reporter.cc +++ b/source/common/upstream/load_stats_reporter.cc @@ -29,6 +29,17 @@ LoadStatsReporter::LoadStatsReporter(const LocalInfo::LocalInfo& local_info, establishNewStream(); } +LoadStatsReporter::~LoadStatsReporter() { + // Disable the timer. + ENVOY_LOG_MISC(info, "Destroying LoadStatsReporter"); + retry_timer_->disableTimer(); + response_timer_->disableTimer(); + if (stream_ != nullptr) { + stream_->resetStream(); + stream_ = nullptr; + } +} + void LoadStatsReporter::setRetryTimer() { ENVOY_LOG(info, "Load reporter stats stream/connection will retry in {} ms.", RETRY_DELAY_MS); retry_timer_->enableTimer(std::chrono::milliseconds(RETRY_DELAY_MS)); diff --git a/source/common/upstream/load_stats_reporter.h b/source/common/upstream/load_stats_reporter.h index eeae2b6050c39..c0485b5cdec9a 100644 --- a/source/common/upstream/load_stats_reporter.h +++ b/source/common/upstream/load_stats_reporter.h @@ -36,6 +36,7 @@ class LoadStatsReporter LoadStatsReporter(const LocalInfo::LocalInfo& local_info, ClusterManager& cluster_manager, Stats::Scope& scope, Grpc::RawAsyncClientSharedPtr&& async_client, Event::Dispatcher& dispatcher); + virtual ~LoadStatsReporter(); // Grpc::AsyncStreamCallbacks void onCreateInitialMetadata(Http::RequestHeaderMap& metadata) override; diff --git a/test/common/upstream/load_stats_reporter_test.cc b/test/common/upstream/load_stats_reporter_test.cc index 949b42a00e426..f7a488da8c567 100644 --- a/test/common/upstream/load_stats_reporter_test.cc +++ b/test/common/upstream/load_stats_reporter_test.cc @@ -37,6 +37,17 @@ class LoadStatsReporterTest : public testing::Test { : retry_timer_(new Event::MockTimer()), response_timer_(new Event::MockTimer()), async_client_(new Grpc::MockAsyncClient()) {} + void TearDown() override { + if (load_stats_reporter_ != nullptr) { + // Validate that LoadStatsReporter correctly shuts down by disabling + // timers and resetting the stream. + EXPECT_CALL(*retry_timer_, disableTimer()); + EXPECT_CALL(*response_timer_, disableTimer()); + EXPECT_CALL(async_stream_, resetStream()); + load_stats_reporter_.reset(); + } + } + void createLoadStatsReporter() { InSequence s; EXPECT_CALL(dispatcher_, createTimer_(_)).WillOnce(Invoke([this](Event::TimerCb timer_cb) { @@ -97,7 +108,6 @@ class LoadStatsReporterTest : public testing::Test { NiceMock cm_; Event::MockDispatcher dispatcher_; Stats::IsolatedStoreImpl stats_store_; - std::unique_ptr load_stats_reporter_; Event::MockTimer* retry_timer_; Event::TimerCb retry_timer_cb_; Event::MockTimer* response_timer_; @@ -105,6 +115,7 @@ class LoadStatsReporterTest : public testing::Test { Grpc::MockAsyncStream async_stream_; Grpc::MockAsyncClient* async_client_; NiceMock local_info_; + std::unique_ptr load_stats_reporter_; }; // Validate that stream creation results in a timer based retry. diff --git a/test/extensions/filters/http/grpc_json_transcoder/grpc_json_transcoder_integration_test.cc b/test/extensions/filters/http/grpc_json_transcoder/grpc_json_transcoder_integration_test.cc index 5a9990bbce34f..d13393560390c 100644 --- a/test/extensions/filters/http/grpc_json_transcoder/grpc_json_transcoder_integration_test.cc +++ b/test/extensions/filters/http/grpc_json_transcoder/grpc_json_transcoder_integration_test.cc @@ -1299,7 +1299,7 @@ TEST_P(GrpcJsonTranscoderIntegrationTest, UTF8) { false); } -TEST_P(GrpcJsonTranscoderIntegrationTest, DisableRequestValidation) { +TEST_P(GrpcJsonTranscoderIntegrationTest, DisableRequestValidationGrpcContentType) { HttpIntegrationTest::initialize(); // Transcoding does not occur from a request with the gRPC content type. @@ -1315,7 +1315,10 @@ TEST_P(GrpcJsonTranscoderIntegrationTest, DisableRequestValidation) { Http::TestResponseHeaderMapImpl{ {":status", "200"}, {"grpc-status", "5"}, {"grpc-message", "Shelf 9999 Not Found"}}, "", true, false, R"({ "theme" : "Children")"); +} +TEST_P(GrpcJsonTranscoderIntegrationTest, DisableRequestValidationUnknownPath) { + HttpIntegrationTest::initialize(); // Transcoding does not occur when unknown path is called. // HTTP Request to is passed directly to gRPC backend. // gRPC response is passed directly to HTTP client. @@ -1328,7 +1331,10 @@ TEST_P(GrpcJsonTranscoderIntegrationTest, DisableRequestValidation) { Http::TestResponseHeaderMapImpl{ {":status", "200"}, {"grpc-status", "5"}, {"grpc-message", "Shelf 9999 Not Found"}}, "", true, false, R"({ "theme" : "Children")"); +} +TEST_P(GrpcJsonTranscoderIntegrationTest, DisableRequestValidationUnknownQueryParam) { + HttpIntegrationTest::initialize(); // Transcoding does not occur when unknown query param is included. // HTTP Request to is passed directly to gRPC backend. // gRPC response is passed directly to HTTP client. diff --git a/test/extensions/filters/udp/dns_filter/dns_filter_integration_test.cc b/test/extensions/filters/udp/dns_filter/dns_filter_integration_test.cc index fde6bf587350f..b9b6d740ae65e 100644 --- a/test/extensions/filters/udp/dns_filter/dns_filter_integration_test.cc +++ b/test/extensions/filters/udp/dns_filter/dns_filter_integration_test.cc @@ -88,6 +88,13 @@ class DnsFilterIntegrationTest : public testing::TestWithParammutable_virtual_hosts(0); auto* route = virtual_host->mutable_routes(0)->mutable_route(); - route->mutable_idle_timeout()->set_seconds(0); - route->mutable_idle_timeout()->set_nanos(200 * 1000 * 1000); + route->mutable_idle_timeout()->set_seconds(1); + route->mutable_idle_timeout()->set_nanos(0); }); initialize(); @@ -850,8 +850,8 @@ TEST_P(WebsocketIntegrationTest, WebSocketUpgradePerTryTimeout) { auto* route_config = hcm.mutable_route_config(); auto* virtual_host = route_config->mutable_virtual_hosts(0); auto* route = virtual_host->mutable_routes(0)->mutable_route(); - route->mutable_retry_policy()->mutable_per_try_timeout()->set_nanos( - 200 * 1000 * 1000); // 200ms per-try timeout + route->mutable_retry_policy()->mutable_per_try_timeout()->set_seconds( + 2); // 2s per-try timeout (safe under msan/tsan) }); initialize(); @@ -886,7 +886,7 @@ TEST_P(WebsocketIntegrationTest, WebSocketUpgradeRouteTimeout) { auto* route_config = hcm.mutable_route_config(); auto* virtual_host = route_config->mutable_virtual_hosts(0); auto* route = virtual_host->mutable_routes(0)->mutable_route(); - route->mutable_timeout()->set_nanos(200 * 1000 * 1000); // 200ms route timeout + route->mutable_timeout()->set_seconds(2); // 2s route timeout (safe under msan/tsan) }); initialize(); @@ -922,7 +922,7 @@ TEST_P(WebsocketIntegrationTest, WebSocketUpgradeRouteTimeoutWithRetries) { auto* route_config = hcm.mutable_route_config(); auto* virtual_host = route_config->mutable_virtual_hosts(0); auto* route = virtual_host->mutable_routes(0)->mutable_route(); - route->mutable_timeout()->set_nanos(200 * 1000 * 1000); // 200ms route timeout + route->mutable_timeout()->set_seconds(2); // 2s route timeout }); initialize(); @@ -936,7 +936,7 @@ TEST_P(WebsocketIntegrationTest, WebSocketUpgradeRouteTimeoutWithRetries) { ASSERT_TRUE(fake_upstreams_[0]->waitForHttpConnection(*dispatcher_, fake_upstream_connection_)); ASSERT_TRUE(fake_upstream_connection_->waitForNewStream(*dispatcher_, upstream_request_)); ASSERT_TRUE(upstream_request_->waitForHeadersComplete()); - upstream_request_->encodeHeaders(upgradeFailedResponseHeaders(), false); + upstream_request_->encodeHeaders(upgradeFailedResponseHeaders(), true); // Wait for the first request to be reset or disconnected ASSERT_TRUE(waitForUpstreamDisconnectOrReset());