diff --git a/changelogs/current/bug_fixes/aws__fixed-signing-deadlock-during-server-init.rst b/changelogs/current/bug_fixes/aws__fixed-signing-deadlock-during-server-init.rst new file mode 100644 index 0000000000000..95b4349d1eb22 --- /dev/null +++ b/changelogs/current/bug_fixes/aws__fixed-signing-deadlock-during-server-init.rst @@ -0,0 +1,3 @@ +Fixed an initialization hang when an extension that gates server initialization +sends an HTTP callout to a cluster whose AWS request signing filter resolves +credentials asynchronously. diff --git a/source/extensions/common/aws/BUILD b/source/extensions/common/aws/BUILD index fe5caa2ce0d65..96b7a48588b01 100644 --- a/source/extensions/common/aws/BUILD +++ b/source/extensions/common/aws/BUILD @@ -134,7 +134,8 @@ envoy_cc_library( ":metadata_fetcher_lib", ":utility_lib", "//envoy/common:time_interface", - "//source/common/common:cancel_wrapper_lib", + "//source/common/common:assert_lib", + "//source/common/common:thread_lib", ], ) diff --git a/source/extensions/common/aws/credential_providers/assume_role_credentials_provider.cc b/source/extensions/common/aws/credential_providers/assume_role_credentials_provider.cc index 2df12fb9d2323..5a3edec2fb3b4 100644 --- a/source/extensions/common/aws/credential_providers/assume_role_credentials_provider.cc +++ b/source/extensions/common/aws/credential_providers/assume_role_credentials_provider.cc @@ -105,7 +105,7 @@ void AssumeRoleCredentialsProvider::continueRefresh() { }; // mark credentials as pending while async completes - credentials_pending_.store(true); + setCredentialsPendingToAllThreads(); metadata_fetcher_->fetch(message, Tracing::NullSpan::instance(), *this); } diff --git a/source/extensions/common/aws/credential_providers/container_credentials_provider.cc b/source/extensions/common/aws/credential_providers/container_credentials_provider.cc index 1572bdbdf1c48..9944bcc988b08 100644 --- a/source/extensions/common/aws/credential_providers/container_credentials_provider.cc +++ b/source/extensions/common/aws/credential_providers/container_credentials_provider.cc @@ -66,7 +66,7 @@ void ContainerCredentialsProvider::refresh() { }; // mark credentials as pending while async completes - credentials_pending_.store(true); + setCredentialsPendingToAllThreads(); metadata_fetcher_->fetch(message, Tracing::NullSpan::instance(), *this); } diff --git a/source/extensions/common/aws/credential_providers/iam_roles_anywhere_credentials_provider.cc b/source/extensions/common/aws/credential_providers/iam_roles_anywhere_credentials_provider.cc index c59516cb9ee23..4fe30a99c8db3 100644 --- a/source/extensions/common/aws/credential_providers/iam_roles_anywhere_credentials_provider.cc +++ b/source/extensions/common/aws/credential_providers/iam_roles_anywhere_credentials_provider.cc @@ -110,7 +110,7 @@ void IAMRolesAnywhereCredentialsProvider::refresh() { }; // mark credentials as pending while async completes - credentials_pending_.store(true); + setCredentialsPendingToAllThreads(); metadata_fetcher_->fetch(message, Tracing::NullSpan::instance(), *this); } diff --git a/source/extensions/common/aws/credential_providers/instance_profile_credentials_provider.cc b/source/extensions/common/aws/credential_providers/instance_profile_credentials_provider.cc index 05a273ba787a9..8b308856fefb6 100644 --- a/source/extensions/common/aws/credential_providers/instance_profile_credentials_provider.cc +++ b/source/extensions/common/aws/credential_providers/instance_profile_credentials_provider.cc @@ -47,7 +47,7 @@ void InstanceProfileCredentialsProvider::refresh() { continue_on_async_fetch_failure_reason_ = "Token fetch failed, falling back to IMDSv1"; // mark credentials as pending while async completes - credentials_pending_.store(true); + setCredentialsPendingToAllThreads(); metadata_fetcher_->fetch(token_req_message, Tracing::NullSpan::instance(), *this); } @@ -71,7 +71,7 @@ void InstanceProfileCredentialsProvider::fetchInstanceRoleAsync(const std::strin }; // mark credentials as pending while async completes - credentials_pending_.store(true); + setCredentialsPendingToAllThreads(); metadata_fetcher_->fetch(message, Tracing::NullSpan::instance(), *this); } @@ -116,7 +116,7 @@ void InstanceProfileCredentialsProvider::fetchCredentialFromInstanceRoleAsync( }; // mark credentials as pending while async completes - credentials_pending_.store(true); + setCredentialsPendingToAllThreads(); metadata_fetcher_->fetch(message, Tracing::NullSpan::instance(), *this); } diff --git a/source/extensions/common/aws/credential_providers/webidentity_credentials_provider.cc b/source/extensions/common/aws/credential_providers/webidentity_credentials_provider.cc index 1bd63b5c08e94..d1261b2f02a37 100644 --- a/source/extensions/common/aws/credential_providers/webidentity_credentials_provider.cc +++ b/source/extensions/common/aws/credential_providers/webidentity_credentials_provider.cc @@ -79,7 +79,7 @@ void WebIdentityCredentialsProvider::refresh() { }; // mark credentials as pending while async completes - credentials_pending_.store(true); + setCredentialsPendingToAllThreads(); metadata_fetcher_->fetch(message, Tracing::NullSpan::instance(), *this); } diff --git a/source/extensions/common/aws/credentials_provider.h b/source/extensions/common/aws/credentials_provider.h index 7ee281882111e..b214acd467836 100644 --- a/source/extensions/common/aws/credentials_provider.h +++ b/source/extensions/common/aws/credentials_provider.h @@ -168,8 +168,11 @@ using CredentialSubscriberCallbacksSharedPtr = std::shared_ptrsubscribeToCredentialUpdates // to register itself for updates via onCredentialUpdate callback. When a credential provider has -// successfully updated all threads with new credentials, via the setCredentialsToAllThreads method -// it will notify all subscribers that credentials have been retrieved. +// posted new credentials to all threads, via the setCredentialsToAllThreads method it will notify +// all subscribers that credentials have been retrieved. +// +// Notification happens as soon as the update has been posted, not once every thread has applied it, +// because the main thread may need credentials before worker threads are running at all. // // Subscription is only relevant for metadata credentials providers, as these are the only // credential providers that implement async credential retrieval functionality. diff --git a/source/extensions/common/aws/metadata_credentials_provider_base.cc b/source/extensions/common/aws/metadata_credentials_provider_base.cc index e3e8039a46189..97cf39217975c 100644 --- a/source/extensions/common/aws/metadata_credentials_provider_base.cc +++ b/source/extensions/common/aws/metadata_credentials_provider_base.cc @@ -4,6 +4,9 @@ #include "envoy/server/factory_context.h" +#include "source/common/common/assert.h" +#include "source/common/common/thread.h" + namespace Envoy { namespace Extensions { namespace Common { @@ -34,7 +37,6 @@ MetadataCredentialsProviderBase::MetadataCredentialsProviderBase( }; MetadataCredentialsProviderBase::~MetadataCredentialsProviderBase() { - cancel_credentials_update_callback_(); if (metadata_fetcher_) { metadata_fetcher_->cancel(); } @@ -68,7 +70,15 @@ void MetadataCredentialsProviderBase::credentialsRetrievalError() { handleFetchDone(); } -bool MetadataCredentialsProviderBase::credentialsPending() { return credentials_pending_; } +bool MetadataCredentialsProviderBase::credentialsPending() { + if (!tls_slot_->currentThreadRegistered()) { + ASSERT(false, "AWS credentials provider queried from a thread with no thread local storage"); + return true; + } + auto cache = tls_slot_->get(); + ASSERT(cache.has_value()); + return !cache.has_value() || cache->credentials_pending_; +} Credentials MetadataCredentialsProviderBase::getCredentials() { return *(*tls_slot_)->credentials_.get(); @@ -140,28 +150,82 @@ void MetadataCredentialsProviderBase::setCredentialsToAllThreads( CredentialsConstSharedPtr shared_credentials = std::move(creds); if (tls_slot_ && !tls_slot_->isShutdown()) { + // A weak_ptr rather than a raw `this`, so that a completion callback still queued when the + // provider goes away becomes a no-op instead of a use-after-free. + std::weak_ptr weak_self = weak_from_this(); + + // Set the credentials and clear the pending flag as a single update, so that no thread can + // observe one without the other. This writes the main thread's slot synchronously and posts the + // same update to every registered worker dispatcher. tls_slot_->runOnAllThreads( - /* Set the credentials */ [shared_credentials]( - OptRef - obj) { obj->credentials_ = shared_credentials; }, - /* Notify waiting signers on completion of credential setting above */ - CancelWrapper::cancelWrapped( - [this]() { - credentials_pending_.store(false); - std::list> subscribers_copy; - { - Thread::LockGuard guard(mu_); - subscribers_copy = credentials_subscribers_; - } - for (auto& weak_cb : subscribers_copy) { - if (auto cb = weak_cb.lock()) { - ENVOY_LOG(debug, "Notifying subscriber of credential update"); - cb->onCredentialUpdate(); - } - } - }, - &cancel_credentials_update_callback_)); + [shared_credentials](OptRef obj) { + obj->credentials_ = shared_credentials; + obj->credentials_pending_ = false; + }, + // Notify a second time once every worker has applied the update. Between the immediate + // notification below and a worker applying its update, that worker still reads + // `credentials_pending_ == true` from its own slot, so it can queue a pending callback + // after the immediate notification has already drained the queue. This notification is + // what wakes such a callback; without it the request stalls until the next successful + // refresh. + [weak_self]() { + if (auto self = weak_self.lock()) { + self->notifySubscribers(); + } + }); + + // Notify waiting signers from this thread as well, rather than relying only on the + // all-threads-complete callback above. That callback does not run until every registered worker + // dispatcher has handled the posted update, and worker dispatchers do not start running until + // `startWorkers()`, which waits on server initialization. The main thread might like to use + // credentials before that point, and waiting for the workers would deadlock. (For example, a + // dynamic modules bootstrap extension might want to make an HTTP callout before it signals + // server init is complete.) + // + // For subscribers that post their wakeup to their own dispatcher (the AWS request signing and + // Lambda filters do), notifying here is ordered correctly: the credential update above is + // posted to each worker first, and dispatcher post queues are FIFO, so a worker applies the + // update before it runs the wakeup that reads it. Subscribers that instead run their callback + // inline rely on this being the main thread, whose slot `runOnAllThreads` has already updated. + // + // Note that unlike the completion callback, this notification runs on the caller's stack, which + // for a credential refresh is inside MetadataFetcher::onSuccess()/onMetadataError() and ahead + // of handleFetchDone(). No subscriber re-enters the provider today, but one that did would see + // a half-finished refresh. + notifySubscribers(); + } +} + +void MetadataCredentialsProviderBase::notifySubscribers() { + std::list> subscribers_copy; + { + Thread::LockGuard guard(mu_); + subscribers_copy = credentials_subscribers_; + } + for (auto& weak_cb : subscribers_copy) { + if (auto cb = weak_cb.lock()) { + ENVOY_LOG(debug, "Notifying subscriber of credential update"); + cb->onCredentialUpdate(); + } + } +} + +void MetadataCredentialsProviderBase::setCredentialsPendingToAllThreads() { + // The dedup below reads the main thread's slot, so it is only meaningful on the main thread. Not + // relying on the assertion inside runOnAllThreads(), because the dedup can return before ever + // reaching it. + ASSERT_IS_MAIN_OR_TEST_THREAD(); + if (!tls_slot_ || tls_slot_->isShutdown()) { + return; + } + // The main thread's slot is written synchronously by runOnAllThreads, so it always holds the most + // recently initiated update. If it already says pending then every other thread has the same + // update applied or queued, and there is nothing to broadcast. + if ((*tls_slot_)->credentials_pending_) { + return; } + tls_slot_->runOnAllThreads( + [](OptRef obj) { obj->credentials_pending_ = true; }); } CredentialSubscriberCallbacksHandlePtr diff --git a/source/extensions/common/aws/metadata_credentials_provider_base.h b/source/extensions/common/aws/metadata_credentials_provider_base.h index 65430d857485c..2eb8a7eae638e 100644 --- a/source/extensions/common/aws/metadata_credentials_provider_base.h +++ b/source/extensions/common/aws/metadata_credentials_provider_base.h @@ -1,6 +1,5 @@ #pragma once -#include "source/common/common/cancel_wrapper.h" #include "source/extensions/common/aws/aws_cluster_manager.h" #include "source/extensions/common/aws/credentials_provider.h" #include "source/extensions/common/aws/metadata_fetcher.h" @@ -61,13 +60,17 @@ class MetadataCredentialsProviderBase subscribeToCredentialUpdates(CredentialSubscriberCallbacksSharedPtr cs); protected: + // Per-thread credential state. Both members are only ever touched on the thread that owns the + // slot: the main thread writes its own copy synchronously from setCredentialsToAllThreads and + // setCredentialsPendingToAllThreads, and every other thread applies the same update from its own + // dispatcher. struct ThreadLocalCredentialsCache : public ThreadLocal::ThreadLocalObject { ThreadLocalCredentialsCache() : credentials_(std::make_shared()) {}; // The credentials object. CredentialsConstSharedPtr credentials_; - // Lock guard. - Thread::MutexBasicLockable lock_; + // Are credentials pending on this thread? + bool credentials_pending_{true}; }; // Set anonymous credentials to all threads, update stats and close async @@ -81,9 +84,18 @@ class MetadataCredentialsProviderBase // Handle fetch done. void handleFetchDone(); - // Set Credentials shared_ptr on all threads. + // Set Credentials shared_ptr on all threads, and mark credentials as no longer pending. void setCredentialsToAllThreads(CredentialsConstUniquePtr&& creds); + // Tell every subscriber that credentials have been retrieved. Called both from the thread that + // initiates a credential update and again once all threads have applied it; see + // setCredentialsToAllThreads. Subscriber notification is idempotent. + void notifySubscribers(); + + // Mark credentials as pending on all threads, ahead of an async credential fetch. Must be called + // on the main thread. + void setCredentialsPendingToAllThreads(); + virtual void refresh() PURE; Server::Configuration::ServerFactoryContext& context_; @@ -122,12 +134,9 @@ class MetadataCredentialsProviderBase AwsClusterManagerPtr aws_cluster_manager_; // RAII handle for callbacks from AWS cluster manager AwsManagedClusterUpdateCallbacksHandlePtr callback_handle_; - // Are credentials pending? - std::atomic credentials_pending_ = true; Thread::MutexBasicLockable mu_; std::list> credentials_subscribers_ ABSL_GUARDED_BY(mu_); - CancelWrapper::CancelFunction cancel_credentials_update_callback_ = []() {}; }; } // namespace Aws diff --git a/test/extensions/common/aws/BUILD b/test/extensions/common/aws/BUILD index b1ac67e22d2ef..5a1d1c456891c 100644 --- a/test/extensions/common/aws/BUILD +++ b/test/extensions/common/aws/BUILD @@ -105,11 +105,14 @@ envoy_cc_test( srcs = ["credentials_provider_test.cc"], rbe_pool = "6gig", deps = [ + "//source/common/thread_local:thread_local_lib", "//source/extensions/common/aws:credentials_provider_interface", "//source/extensions/common/aws:utility_lib", "//source/extensions/common/aws/signers:sigv4_signer_impl_lib", "//test/extensions/common/aws:aws_mocks", "//test/mocks/server:factory_context_mocks", "//test/test_common:status_utility_lib", + "//test/test_common:utility_lib", + "@abseil-cpp//absl/synchronization", ], ) diff --git a/test/extensions/common/aws/credential_providers/assume_role_credentials_provider_test.cc b/test/extensions/common/aws/credential_providers/assume_role_credentials_provider_test.cc index d515134eae286..e88823b28ef71 100644 --- a/test/extensions/common/aws/credential_providers/assume_role_credentials_provider_test.cc +++ b/test/extensions/common/aws/credential_providers/assume_role_credentials_provider_test.cc @@ -973,8 +973,10 @@ TEST_F(AssumeRoleCredentialsProviderTest, WithExternalId) { EXPECT_EQ("test-access-key", credentials.accessKeyId().value()); } -// Tests ASAN failure when cancel wrapper is not used -TEST_F(AssumeRoleCredentialsProviderTest, CancelWrapperPreventsUseAfterFree) { +// setCredentialsToAllThreads() leaves an all-threads-complete callback outstanding, which may run +// after the provider is gone. It captures a weak_ptr, so firing it then must be a no-op rather than +// a use-after-free (this test fails under ASAN if the callback captures the provider directly). +TEST_F(AssumeRoleCredentialsProviderTest, DeferredCompletionCallbackSafeAfterDestruction) { std::function captured_callback; EXPECT_CALL(context_.thread_local_, runOnAllThreads(testing::_, testing::_)) diff --git a/test/extensions/common/aws/credential_providers/instance_profile_credentials_provider_test.cc b/test/extensions/common/aws/credential_providers/instance_profile_credentials_provider_test.cc index a034b8b689413..ceff72bc232d0 100644 --- a/test/extensions/common/aws/credential_providers/instance_profile_credentials_provider_test.cc +++ b/test/extensions/common/aws/credential_providers/instance_profile_credentials_provider_test.cc @@ -769,8 +769,10 @@ not json delete (raw_metadata_fetcher_); } -// Tests ASAN failure when cancel wrapper is not used -TEST_F(InstanceProfileCredentialsProviderTest, CancelWrapperPreventsUseAfterFree) { +// setCredentialsToAllThreads() leaves an all-threads-complete callback outstanding, which may run +// after the provider is gone. It captures a weak_ptr, so firing it then must be a no-op rather than +// a use-after-free (this test fails under ASAN if the callback captures the provider directly). +TEST_F(InstanceProfileCredentialsProviderTest, DeferredCompletionCallbackSafeAfterDestruction) { std::function captured_callback; EXPECT_CALL(context_.thread_local_, runOnAllThreads(testing::_, testing::_)) diff --git a/test/extensions/common/aws/credentials_provider_test.cc b/test/extensions/common/aws/credentials_provider_test.cc index 34f00df16b847..f5d1c65e095c7 100644 --- a/test/extensions/common/aws/credentials_provider_test.cc +++ b/test/extensions/common/aws/credentials_provider_test.cc @@ -1,4 +1,5 @@ #include "source/common/http/message_impl.h" +#include "source/common/thread_local/thread_local_impl.h" #include "source/extensions/common/aws/credentials_provider.h" #include "source/extensions/common/aws/signers/sigv4_signer_impl.h" @@ -6,12 +7,15 @@ #include "test/mocks/event/mocks.h" #include "test/mocks/server/server_factory_context.h" #include "test/test_common/status_utility.h" +#include "test/test_common/utility.h" +#include "absl/synchronization/notification.h" #include "gtest/gtest.h" using Envoy::Extensions::Common::Aws::MetadataFetcherPtr; using testing::MockFunction; using testing::Return; +using testing::ReturnRef; namespace Envoy { namespace Extensions { @@ -163,8 +167,10 @@ TEST_F(AsyncCredentialHandlingTest, ChainCallbackCalledWhenCredentialsReturned) timer_ = new NiceMock(&context_.dispatcher_); timer_->enableTimer(std::chrono::milliseconds(1), nullptr); + // Subscribers are notified once from the thread that initiates the update and again once + // every thread has applied it. ThreadLocal::MockInstance runs both synchronously. auto chain = std::make_shared(); - EXPECT_CALL(*chain, onCredentialUpdate()); + EXPECT_CALL(*chain, onCredentialUpdate()).Times(2); EXPECT_CALL(*chain, chainGetCredentials()).WillRepeatedly(Return(Credentials("akid", "skid"))); auto document = R"EOF( @@ -351,8 +357,10 @@ TEST_F(AsyncCredentialHandlingTest, SubscriptionsCleanedUp) { timer_ = new NiceMock(&context_.dispatcher_); timer_->enableTimer(std::chrono::milliseconds(1), nullptr); + // Subscribers are notified once from the thread that initiates the update and again once + // every thread has applied it. ThreadLocal::MockInstance runs both synchronously. auto chain = std::make_shared(); - EXPECT_CALL(*chain, onCredentialUpdate()); + EXPECT_CALL(*chain, onCredentialUpdate()).Times(2); EXPECT_CALL(*chain, chainGetCredentials()).WillRepeatedly(Return(Credentials("akid", "skid"))); auto chain2 = std::make_shared(); @@ -509,9 +517,10 @@ TEST_F(AsyncCredentialHandlingTest, WeakPtrProtectionInSubscriberCallback) { auto provider_friend = MetadataCredentialsProviderBaseFriend(provider_); - // Test 1: When subscriber is alive, onCredentialUpdate should be called + // Test 1: When subscriber is alive, onCredentialUpdate should be called. It is called once from + // the updating thread and again once all threads have applied the update. auto chain = std::make_shared(); - EXPECT_CALL(*chain, onCredentialUpdate()); + EXPECT_CALL(*chain, onCredentialUpdate()).Times(2); auto handle = provider_->subscribeToCredentialUpdates(chain); // Trigger credential update @@ -649,6 +658,311 @@ TEST(CredentialsProviderChainTest, CheckChainReturnsPendingInCorrectOrder) { EXPECT_EQ(creds.secretAccessKey(), "1"); } +// Regression tests for a deadlock between credential resolution and worker thread startup. +// +// credentials_pending_ used to be cleared from the all-threads-complete callback of +// ThreadLocal::Slot::runOnAllThreads(). That callback only fires once every registered worker +// dispatcher has run the update, and worker dispatchers are registered when the ListenerManager is +// constructed but do not start running until startWorkers(), which itself waits on the server init +// manager. Any credential refresh that resolved before that point latched credentials_pending_ at +// true forever and stalled every signer depending on it. +// +// These tests use a real ThreadLocal::InstanceImpl with a registered worker dispatcher that is +// deliberately never run(), which is exactly the pre-startWorkers() state. Every other AWS test +// uses ThreadLocal::MockInstance, whose runOnAllThreads() invokes both callbacks synchronously, so +// the barrier is never real there. +class CredentialsPendingBeforeWorkerStartupTest : public testing::Test { +public: + CredentialsPendingBeforeWorkerStartupTest() + : api_(Api::createApiForTest()), main_dispatcher_(api_->allocateDispatcher("test_main")), + worker_dispatcher_(api_->allocateDispatcher("test_worker")), + mock_manager_(std::make_shared()) { + tls_.registerThread(*main_dispatcher_, /*main_thread=*/true); + // Registered but never run(), modelling a worker thread that has not been started yet. + tls_.registerThread(*worker_dispatcher_, /*main_thread=*/false); + + // Must be in place before the provider is constructed: MetadataCredentialsProviderBase's + // constructor allocates its tls slot from context_.threadLocal(). + ON_CALL(context_, threadLocal()).WillByDefault(ReturnRef(tls_)); + ON_CALL(*mock_manager_, getUriFromClusterName(_)).WillByDefault(Return("uri_2")); + } + + ~CredentialsPendingBeforeWorkerStartupTest() override { + // The provider owns a slot in tls_, so it has to go away while tls_ is still alive and not yet + // shut down. + provider_.reset(); + tls_.shutdownGlobalThreading(); + tls_.shutdownThread(); + } + + void createProvider() { + envoy::extensions::common::aws::v3::AssumeRoleWithWebIdentityCredentialProvider cred_provider = + {}; + cred_provider.mutable_web_identity_token_data_source()->set_inline_string("abced"); + cred_provider.set_role_arn("aws:iam::123456789012:role/arn"); + cred_provider.set_role_session_name("role-session-name"); + + // These tests drive setCredentialsToAllThreads() directly rather than through a fetch, so the + // metadata fetcher is never created. + provider_ = std::make_shared( + context_, mock_manager_, "cluster_2", + [](Upstream::ClusterManager&, absl::string_view) { return MetadataFetcherPtr{}; }, + MetadataFetcher::MetadataReceiver::RefreshState::Ready, std::chrono::seconds(2), + cred_provider); + } + + Api::ApiPtr api_; + Event::DispatcherPtr main_dispatcher_; + Event::DispatcherPtr worker_dispatcher_; + ThreadLocal::InstanceImpl tls_; + NiceMock context_; + std::shared_ptr mock_manager_; + WebIdentityCredentialsProviderPtr provider_; +}; + +// A successful credential refresh must un-pend the provider and notify its subscribers even though +// no worker dispatcher is running. +TEST_F(CredentialsPendingBeforeWorkerStartupTest, UnpendsBeforeWorkerDispatchersRun) { + createProvider(); + EXPECT_TRUE(provider_->credentialsPending()); + + auto chain = std::make_shared(); + EXPECT_CALL(*chain, onCredentialUpdate()); + auto handle = provider_->subscribeToCredentialUpdates(chain); + + MetadataCredentialsProviderBaseFriend(provider_).setCredentialsToAllThreads( + std::make_unique("akid", "secret", "token")); + + EXPECT_FALSE(provider_->credentialsPending()); + // The main thread's slot is written synchronously by runOnAllThreads(), so the credentials are + // readable here as well. + const auto credentials = provider_->getCredentials(); + EXPECT_EQ("akid", credentials.accessKeyId()); + EXPECT_EQ("secret", credentials.secretAccessKey()); + EXPECT_EQ("token", credentials.sessionToken()); +} + +// The failure path must un-pend too. +TEST_F(CredentialsPendingBeforeWorkerStartupTest, UnpendsOnRetrievalErrorBeforeWorkersRun) { + createProvider(); + EXPECT_TRUE(provider_->credentialsPending()); + + auto chain = std::make_shared(); + EXPECT_CALL(*chain, onCredentialUpdate()); + auto handle = provider_->subscribeToCredentialUpdates(chain); + + MetadataCredentialsProviderBaseFriend(provider_).credentialsRetrievalError(); + + EXPECT_FALSE(provider_->credentialsPending()); + EXPECT_FALSE(provider_->getCredentials().accessKeyId().has_value()); +} + +// A thread must never observe "credentials are no longer pending" before the credential update +// itself has been applied to that thread. The pending flag and the credentials both live in the +// thread local cache and move as one update, so the pair is always consistent per thread. +// +// This test parks a running worker dispatcher inside one of its own callbacks, performs a +// credential update from the main thread while the worker is stuck there, and then lets the worker +// observe both halves. +class CredentialsPendingWorkerVisibilityTest : public testing::Test { +public: + CredentialsPendingWorkerVisibilityTest() + : api_(Api::createApiForTest()), main_dispatcher_(api_->allocateDispatcher("test_main")), + worker_dispatcher_(api_->allocateDispatcher("test_worker")), + mock_manager_(std::make_shared()) { + tls_.registerThread(*main_dispatcher_, /*main_thread=*/true); + tls_.registerThread(*worker_dispatcher_, /*main_thread=*/false); + + ON_CALL(context_, threadLocal()).WillByDefault(ReturnRef(tls_)); + ON_CALL(*mock_manager_, getUriFromClusterName(_)).WillByDefault(Return("uri_2")); + } + + ~CredentialsPendingWorkerVisibilityTest() override { + if (worker_thread_ != nullptr) { + worker_dispatcher_->exit(); + worker_thread_->join(); + } + provider_.reset(); + tls_.shutdownGlobalThreading(); + tls_.shutdownThread(); + } + + void createProvider() { + envoy::extensions::common::aws::v3::AssumeRoleWithWebIdentityCredentialProvider cred_provider = + {}; + cred_provider.mutable_web_identity_token_data_source()->set_inline_string("abced"); + cred_provider.set_role_arn("aws:iam::123456789012:role/arn"); + cred_provider.set_role_session_name("role-session-name"); + + provider_ = std::make_shared( + context_, mock_manager_, "cluster_2", + [](Upstream::ClusterManager&, absl::string_view) { return MetadataFetcherPtr{}; }, + MetadataFetcher::MetadataReceiver::RefreshState::Ready, std::chrono::seconds(2), + cred_provider); + } + + // Started after the provider exists, so the slot initializer that the provider's constructor + // posted is already queued ahead of anything this test posts. RunUntilExit is what production + // worker threads use: RunType::Block returns as soon as the event set drains, which for a + // dispatcher with no listeners on it means the thread would exit between posts. + void startWorker() { + worker_thread_ = api_->threadFactory().createThread( + [this]() { worker_dispatcher_->run(Event::Dispatcher::RunType::RunUntilExit); }); + } + + // Runs cb on the worker thread and waits for it to finish. + void runOnWorker(std::function cb) { + absl::Notification done; + worker_dispatcher_->post([&]() { + cb(); + done.Notify(); + }); + done.WaitForNotification(); + } + + Api::ApiPtr api_; + Event::DispatcherPtr main_dispatcher_; + Event::DispatcherPtr worker_dispatcher_; + Thread::ThreadPtr worker_thread_; + ThreadLocal::InstanceImpl tls_; + NiceMock context_; + std::shared_ptr mock_manager_; + WebIdentityCredentialsProviderPtr provider_; +}; + +TEST_F(CredentialsPendingWorkerVisibilityTest, WorkerNeverSeesUnpendedStaleCredentials) { + createProvider(); + startWorker(); + + absl::Notification worker_parked; + absl::Notification release_worker; + absl::Notification observation_done; + bool observed_pending = false; + bool observed_has_credentials = false; + + // Occupies the worker's event loop, so the credential update posted below is queued behind this + // callback and cannot be applied until it returns. + worker_dispatcher_->post([&]() { + worker_parked.Notify(); + release_worker.WaitForNotification(); + observed_pending = provider_->credentialsPending(); + observed_has_credentials = provider_->getCredentials().hasCredentials(); + observation_done.Notify(); + }); + worker_parked.WaitForNotification(); + + MetadataCredentialsProviderBaseFriend(provider_).setCredentialsToAllThreads( + std::make_unique("akid", "secret", "token")); + // Visible on the main thread immediately, which is what lets a callout made before startWorkers() + // be signed. + EXPECT_FALSE(provider_->credentialsPending()); + EXPECT_TRUE(provider_->getCredentials().hasCredentials()); + + release_worker.Notify(); + observation_done.WaitForNotification(); + + // The worker had not applied the update yet, so it has to still report pending. + EXPECT_TRUE(observed_pending); + EXPECT_FALSE(observed_has_credentials); + + // Once it drains its post queue, both halves are visible together. + bool drained_pending = true; + bool drained_has_credentials = false; + runOnWorker([&]() { + drained_pending = provider_->credentialsPending(); + drained_has_credentials = provider_->getCredentials().hasCredentials(); + }); + EXPECT_FALSE(drained_pending); + EXPECT_TRUE(drained_has_credentials); +} + +// A pending callback registered by a worker after the updating thread has already notified +// subscribers must still be woken. The worker keeps reading credentials_pending_ == true from its +// own slot until it applies the posted update, so it can queue a callback onto a queue that has +// just been drained. The all-threads-complete notification is what rescues it; without it the +// request stalls until the next successful refresh, ~1 hour later. +TEST_F(CredentialsPendingWorkerVisibilityTest, PendingCallbackRegisteredDuringStaleWindowIsWoken) { + createProvider(); + startWorker(); + + auto chain = std::make_shared(); + chain->add(provider_); + auto handle = provider_->subscribeToCredentialUpdates(chain); + + absl::Notification worker_parked; + absl::Notification release_worker; + absl::Notification registered; + bool was_pending = false; + std::atomic callback_fired{false}; + + // Parks the worker's event loop so that the credential update posted below cannot be applied to + // the worker's slot until we let it go. + worker_dispatcher_->post([&]() { + worker_parked.Notify(); + release_worker.WaitForNotification(); + // The worker's slot still says pending, so the chain queues this callback - after the main + // thread has already drained the queue. + was_pending = + chain->addCallbackIfChainCredentialsPending([&callback_fired]() { callback_fired = true; }); + registered.Notify(); + }); + worker_parked.WaitForNotification(); + + MetadataCredentialsProviderBaseFriend(provider_).setCredentialsToAllThreads( + std::make_unique("akid", "secret", "token")); + + release_worker.Notify(); + registered.WaitForNotification(); + EXPECT_TRUE(was_pending); + EXPECT_FALSE(callback_fired); + + // Let the worker drain the credential update. Releasing the last reference to the update callback + // posts the all-threads-complete callback onto the main dispatcher. + runOnWorker([]() {}); + main_dispatcher_->run(Event::Dispatcher::RunType::NonBlock); + + EXPECT_TRUE(callback_fired); +} + +// Marking pending again while already pending must not broadcast. +// InstanceProfileCredentialsProvider marks pending once per stage of its three stage fetch, and +// only the first transition needs an update posted to every thread. +TEST(MetadataCredentialsProviderPendingTest, MarkingPendingIsDeduplicated) { + NiceMock context; + auto mock_manager = std::make_shared(); + ON_CALL(*mock_manager, getUriFromClusterName(_)).WillByDefault(Return("uri_2")); + + envoy::extensions::common::aws::v3::AssumeRoleWithWebIdentityCredentialProvider cred_provider = + {}; + cred_provider.mutable_web_identity_token_data_source()->set_inline_string("abced"); + cred_provider.set_role_arn("aws:iam::123456789012:role/arn"); + cred_provider.set_role_session_name("role-session-name"); + + auto provider = std::make_shared( + context, mock_manager, "cluster_2", + [](Upstream::ClusterManager&, absl::string_view) { return MetadataFetcherPtr{}; }, + MetadataFetcher::MetadataReceiver::RefreshState::Ready, std::chrono::seconds(2), + cred_provider); + auto provider_friend = MetadataCredentialsProviderBaseFriend(provider); + + // A provider starts out pending, so there is nothing to broadcast. + EXPECT_CALL(context.thread_local_, runOnAllThreads(_)).Times(0); + provider_friend.setCredentialsPendingToAllThreads(); + EXPECT_TRUE(provider->credentialsPending()); + testing::Mock::VerifyAndClearExpectations(&context.thread_local_); + + // Delivering the credentials and clearing the flag uses the two-argument overload, so that + // subscribers are notified again once every thread has applied the update. Setting the flag again + // is then the only single-argument broadcast: the second, redundant call posts nothing. + EXPECT_CALL(context.thread_local_, runOnAllThreads(_, _)); + EXPECT_CALL(context.thread_local_, runOnAllThreads(_)); + provider_friend.setCredentialsToAllThreads(std::make_unique("akid", "secret")); + EXPECT_FALSE(provider->credentialsPending()); + provider_friend.setCredentialsPendingToAllThreads(); + provider_friend.setCredentialsPendingToAllThreads(); + EXPECT_TRUE(provider->credentialsPending()); +} + } // namespace Aws } // namespace Common } // namespace Extensions diff --git a/test/extensions/common/aws/mocks.h b/test/extensions/common/aws/mocks.h index cb2a9f26ae839..9f015f767c64b 100644 --- a/test/extensions/common/aws/mocks.h +++ b/test/extensions/common/aws/mocks.h @@ -163,6 +163,8 @@ class MetadataCredentialsProviderBaseFriend { void setCredentialsToAllThreads(CredentialsConstUniquePtr&& creds) { provider_->setCredentialsToAllThreads(std::move(creds)); } + void setCredentialsPendingToAllThreads() { provider_->setCredentialsPendingToAllThreads(); } + void credentialsRetrievalError() { provider_->credentialsRetrievalError(); } void invalidateStats() { provider_->stats_.reset(); } size_t getSubscribersCount() { Thread::LockGuard lock(provider_->mu_); diff --git a/test/extensions/dynamic_modules/bootstrap/BUILD b/test/extensions/dynamic_modules/bootstrap/BUILD index e83e7a4668abb..398d5fa016f2d 100644 --- a/test/extensions/dynamic_modules/bootstrap/BUILD +++ b/test/extensions/dynamic_modules/bootstrap/BUILD @@ -129,11 +129,14 @@ envoy_cc_test( "//test/extensions/dynamic_modules/test_data/rust:bootstrap_integration_test", "//test/extensions/dynamic_modules/test_data/rust:bootstrap_listener_lifecycle_test", "//test/extensions/dynamic_modules/test_data/rust:bootstrap_shared_data_test", + "//test/extensions/dynamic_modules/test_data/rust:bootstrap_signed_callout_test", "//test/extensions/dynamic_modules/test_data/rust:bootstrap_stats_test", "//test/extensions/dynamic_modules/test_data/rust:bootstrap_timer_test", ], deps = [ + "//source/common/router:upstream_codec_filter_lib", "//source/extensions/bootstrap/dynamic_modules:config", + "//source/extensions/filters/http/aws_request_signing:config", "//source/extensions/filters/http/dynamic_modules:factory_registration", "//test/integration:http_integration_lib", "//test/test_common:environment_lib", diff --git a/test/extensions/dynamic_modules/bootstrap/integration_test.cc b/test/extensions/dynamic_modules/bootstrap/integration_test.cc index b6e6e9044f3d9..f277b79b077df 100644 --- a/test/extensions/dynamic_modules/bootstrap/integration_test.cc +++ b/test/extensions/dynamic_modules/bootstrap/integration_test.cc @@ -316,6 +316,67 @@ name: envoy.extensions.filters.http.dynamic_modules } } +const std::string AWS_REQUEST_SIGNING_UPSTREAM_FILTER = R"EOF( +name: envoy.filters.http.aws_request_signing +typed_config: + "@type": type.googleapis.com/envoy.extensions.filters.http.aws_request_signing.v3.AwsRequestSigning + service_name: execute-api + region: us-east-1 + signing_algorithm: aws_sigv4 + credential_provider: + custom_credential_provider_chain: true + container_credential_provider: {} +)EOF"; + +class DynamicModulesBootstrapAwsSigningIntegrationTest + : public DynamicModulesBootstrapIntegrationTest { +public: + DynamicModulesBootstrapAwsSigningIntegrationTest() { + // Nothing is listening here, so the fetch fails, anonymous credentials are installed, and the + // callout goes out unsigned. That doesn't matter. The deadlock is in clearing the pending flag, + // not in the signature, and both the success and the failure path of the fetch reach it + // through setCredentialsToAllThreads(). + TestEnvironment::setEnvVar("AWS_CONTAINER_CREDENTIALS_FULL_URI", + "http://127.0.0.1:1/path/to/creds", 1); + } + + ~DynamicModulesBootstrapAwsSigningIntegrationTest() override { + // Undo environment changes. + TestEnvironment::unsetEnvVar("AWS_CONTAINER_CREDENTIALS_FULL_URI"); + } +}; + +INSTANTIATE_TEST_SUITE_P(IpVersions, DynamicModulesBootstrapAwsSigningIntegrationTest, + testing::ValuesIn(TestEnvironment::getIpVersionsForTest()), + TestUtility::ipTestParamsToString); + +// Regression test for a deadlock between the bootstrap init target and AWS credential resolution. +// The module holds its init target open until an HTTP callout through cluster_0 completes, and +// cluster_0 carries an upstream aws_request_signing filter whose credentials chain has no +// synchronous provider, so signing has to wait on an async metadata fetch. +// +// That fetch resolves on the main thread before any worker thread exists. While the resulting +// "credentials are no longer pending" notification was driven from the all-threads-complete +// callback of runOnAllThreads(), it could never fire here: that callback waits for every +// registered worker dispatcher to run the update, worker dispatchers do not run until +// startWorkers(), and startWorkers() waits on the very init manager this module is holding open. +// The signing filter then held the callout until its deadline. Reaching the module's success log at +// all is the assertion; a regression instead fails on the harness giving up waiting for listeners, +// and the module budgets its retries so it cannot spin indefinitely behind that. +TEST_P(DynamicModulesBootstrapAwsSigningIntegrationTest, SignedCalloutGatingInitTarget) { + // Nothing is servicing the fake upstream while the server is still initializing, so it has to + // answer the callout on its own. + autonomous_upstream_ = true; + config_helper_.prependFilter(AWS_REQUEST_SIGNING_UPSTREAM_FILTER, /*downstream=*/false); + // cluster_0 carries no protocol options in the base config, and upstream_protocol_options is a + // required field once any are set. This fills it in without disturbing the filter chain above. + setUpstreamProtocol(Http::CodecType::HTTP1); + + EXPECT_LOG_CONTAINS( + "info", "Bootstrap signed callout test completed successfully!", + initializeWithBootstrapExtension(testDataDir("rust"), "bootstrap_signed_callout_test")); +} + } // namespace DynamicModules } // namespace Bootstrap } // namespace Extensions diff --git a/test/extensions/dynamic_modules/test_data/rust/BUILD b/test/extensions/dynamic_modules/test_data/rust/BUILD index 3bb811cdf6ba5..316fa652fd9ab 100644 --- a/test/extensions/dynamic_modules/test_data/rust/BUILD +++ b/test/extensions/dynamic_modules/test_data/rust/BUILD @@ -49,6 +49,8 @@ test_program(name = "bootstrap_init_target_test") test_program(name = "bootstrap_timer_test") +test_program(name = "bootstrap_signed_callout_test") + test_program(name = "bootstrap_file_watcher_test") test_program(name = "bootstrap_admin_handler_test") diff --git a/test/extensions/dynamic_modules/test_data/rust/Cargo.toml b/test/extensions/dynamic_modules/test_data/rust/Cargo.toml index 320980f29cd54..4b2e3e957dcbb 100644 --- a/test/extensions/dynamic_modules/test_data/rust/Cargo.toml +++ b/test/extensions/dynamic_modules/test_data/rust/Cargo.toml @@ -74,6 +74,12 @@ path = "bootstrap_timer_test.rs" crate-type = ["cdylib"] test = true +[[example]] +name = "bootstrap_signed_callout_test" +path = "bootstrap_signed_callout_test.rs" +crate-type = ["cdylib"] +test = true + [[example]] name = "bootstrap_admin_handler_test" path = "bootstrap_admin_handler_test.rs" diff --git a/test/extensions/dynamic_modules/test_data/rust/bootstrap_signed_callout_test.rs b/test/extensions/dynamic_modules/test_data/rust/bootstrap_signed_callout_test.rs new file mode 100644 index 0000000000000..addcc11f4b168 --- /dev/null +++ b/test/extensions/dynamic_modules/test_data/rust/bootstrap_signed_callout_test.rs @@ -0,0 +1,214 @@ +//! Test module reproducing the shape of a bootstrap extension that gates server initialization on +//! an AWS-signed HTTP callout. +//! +//! The module holds its init target open from `new_bootstrap_extension_config` and only signals +//! completion once an HTTP callout through `cluster_0` has returned a 2xx. The callout is issued +//! with a request body so that an upstream `aws_request_signing` filter on that cluster takes its +//! `decodeData` signing path, which is where a request blocks while AWS credentials are still +//! pending. +//! +//! Combined with a credentials chain that has no synchronous provider, this closes a loop that +//! used to deadlock: signing waited on an async credential refresh, that refresh could not +//! resolve until worker threads were running, and worker threads could not start until this init +//! target completed. +//! +//! Failures are retried rather than given up on immediately, so that a slow environment does not +//! turn into a flake, but the retries are budgeted so the module cannot loop forever. Once the +//! budget is exhausted it signals init complete without logging its success message, which fails +//! the integration test's log assertion. (A regression is bounded either way, because the test +//! harness gives up waiting for listeners after 20 seconds.) + +use envoy_proxy_dynamic_modules_rust_sdk::*; +use std::sync::atomic::{AtomicBool, AtomicU32, Ordering}; +use std::sync::Arc; +use std::time::Duration; + +declare_bootstrap_init_functions!(my_program_init, my_new_bootstrap_extension_config_fn); + +/// Cluster the callout is sent to. This is the default static cluster of the integration test, +/// which carries the upstream `aws_request_signing` filter. +const CALLOUT_CLUSTER: &str = "cluster_0"; + +/// Callout deadline. Generous relative to the credential refresh it waits behind, so that a +/// timeout here means the request never got signed rather than that it was merely slow. +const CALLOUT_TIMEOUT_MS: u64 = 5000; + +/// Delay before the first callout attempt and between retries. +const CALLOUT_DELAY: Duration = Duration::from_millis(100); + +/// How many callouts may complete unsuccessfully before the module stops retrying. With the bug +/// this reproduces every attempt burns the full `CALLOUT_TIMEOUT_MS`, so this keeps the total under +/// half a minute. +const MAX_CALLOUT_FAILURES: u32 = 4; + +/// How many times the callout may fail to start, which happens while `cluster_0` is still warming +/// up. These retries cost only `CALLOUT_DELAY` each, so the budget is larger. +const MAX_START_FAILURES: u32 = 50; + +/// Scheduler event id used to hop from `on_server_initialized` back onto the config. +const EVENT_SERVER_INITIALIZED: u64 = 1; + +fn my_program_init() -> bool { + true +} + +fn my_new_bootstrap_extension_config_fn( + envoy_extension_config: &mut dyn EnvoyBootstrapExtensionConfig, + _name: &str, + _config: &[u8], +) -> Option> { + // Deliberately do NOT signal init complete here. The init target stays open until the callout + // below succeeds, which is the whole point of this module. + let timer = envoy_extension_config.new_timer(); + let scheduler = Arc::new(envoy_extension_config.new_scheduler()); + envoy_log_info!("signed callout module: config created, init target held open"); + Some(Box::new(SignedCalloutConfig { + timer, + scheduler, + init_signaled: AtomicBool::new(false), + callout_failures: AtomicU32::new(0), + start_failures: AtomicU32::new(0), + })) +} + +struct SignedCalloutConfig { + timer: Box, + scheduler: Arc>, + init_signaled: AtomicBool, + callout_failures: AtomicU32, + start_failures: AtomicU32, +} + +impl SignedCalloutConfig { + /// Signals init complete at most once, so a retry that races a success cannot double-signal. + /// Returns whether this call was the one that signalled. + fn signal_init_complete_once( + &self, + envoy_extension_config: &mut dyn EnvoyBootstrapExtensionConfig, + ) -> bool { + let signalled = self + .init_signaled + .compare_exchange(false, true, Ordering::SeqCst, Ordering::SeqCst) + .is_ok(); + if signalled { + envoy_extension_config.signal_init_complete(); + } + signalled + } + + /// Unblocks server initialization without logging the success message. The integration test + /// asserts on that message, so it fails on the assertion rather than waiting out the test + /// timeout. + fn give_up(&self, envoy_extension_config: &mut dyn EnvoyBootstrapExtensionConfig, reason: &str) { + envoy_log_info!("signed callout module: giving up, {}", reason); + self.signal_init_complete_once(envoy_extension_config); + } +} + +impl BootstrapExtensionConfig for SignedCalloutConfig { + fn new_bootstrap_extension( + &self, + _envoy_extension: &mut dyn EnvoyBootstrapExtension, + ) -> Box { + Box::new(SignedCalloutExtension { + scheduler: self.scheduler.clone(), + }) + } + + fn on_scheduled( + &self, + _envoy_extension_config: &mut dyn EnvoyBootstrapExtensionConfig, + event_id: u64, + ) { + if event_id == EVENT_SERVER_INITIALIZED { + self.timer.enable(CALLOUT_DELAY); + } + } + + fn on_timer_fired( + &self, + envoy_extension_config: &mut dyn EnvoyBootstrapExtensionConfig, + _timer: &dyn EnvoyBootstrapExtensionTimer, + ) { + // A body is sent so that the upstream aws_request_signing filter signs from decodeData + // (use_unsigned_payload defaults to false). + let (result, callout_id) = envoy_extension_config.send_http_callout( + CALLOUT_CLUSTER, + vec![ + (":method", b"POST"), + (":path", b"/oauth2/token"), + (":authority", b"authorizer"), + ], + Some(b"grant_type=client_credentials"), + CALLOUT_TIMEOUT_MS, + ); + if result != abi::envoy_dynamic_module_type_http_callout_init_result::Success { + // The cluster may not be warm yet on the first attempt; keep retrying within budget. + if self.start_failures.fetch_add(1, Ordering::SeqCst) + 1 > MAX_START_FAILURES { + self.give_up(envoy_extension_config, "callout never started"); + return; + } + envoy_log_info!("signed callout module: callout could not be started, retrying"); + self.timer.enable(CALLOUT_DELAY); + return; + } + envoy_log_info!("signed callout module: callout {} in flight", callout_id); + } + + fn on_http_callout_done( + &self, + envoy_extension_config: &mut dyn EnvoyBootstrapExtensionConfig, + _callout_id: u64, + result: abi::envoy_dynamic_module_type_http_callout_result, + response_headers: Option<&[(EnvoyBuffer, EnvoyBuffer)]>, + _response_body: Option<&[EnvoyBuffer]>, + ) { + let status = if result == abi::envoy_dynamic_module_type_http_callout_result::Success { + response_headers + .unwrap_or_default() + .iter() + .find(|(name, _)| name.as_slice() == b":status") + .and_then(|(_, value)| std::str::from_utf8(value.as_slice()).ok()) + .and_then(|value| value.parse::().ok()) + .unwrap_or(0) + } else { + 0 + }; + + if (200..300).contains(&status) { + if self.signal_init_complete_once(envoy_extension_config) { + envoy_log_info!("Bootstrap signed callout test completed successfully!"); + } + return; + } + + // Anything else -- a reset, or the 504 the router emits when signing never completes and the + // callout deadline expires -- is retried within budget, so a slow environment does not flake. + if self.callout_failures.fetch_add(1, Ordering::SeqCst) + 1 > MAX_CALLOUT_FAILURES { + self.give_up( + envoy_extension_config, + "callout never returned a successful status", + ); + return; + } + envoy_log_info!( + "signed callout module: callout failed with status {}, retrying", + status + ); + self.timer.enable(CALLOUT_DELAY); + } +} + +struct SignedCalloutExtension { + scheduler: Arc>, +} + +impl BootstrapExtension for SignedCalloutExtension { + fn on_server_initialized(&mut self, _envoy_extension: &mut dyn EnvoyBootstrapExtension) { + // The cluster manager is only reachable from the config once the server is initialized, so + // the callout cannot be issued any earlier than this. Hop back onto the config to arm the + // timer that sends it. + envoy_log_info!("signed callout module: server initialized, scheduling callout"); + self.scheduler.commit(EVENT_SERVER_INITIALIZED); + } +}