Conversation
Shutdown() never read its timeout parameter and always returned true regardless of whether anything had actually flushed. It now flushes pending exports against the caller's deadline via ForceFlush(timeout) before cancelling sessions, and returns what that flush reported. Also closes an admission race: Export() could pass its isShutdown() check and still register a session after Shutdown() had already taken its session_counter_ snapshot in ForceFlush(), so the flush could return without ever having waited for it. Both the shutdown flag and session registration now share one lock. Fixes open-telemetry#4359
Codecov Report✅ All modified and coverable lines are covered by tests. Additional details and impacted files@@ Coverage Diff @@
## main #4523 +/- ##
==========================================
+ Coverage 86.51% 86.66% +0.15%
==========================================
Files 525 525
Lines 20464 20489 +25
==========================================
+ Hits 17702 17754 +52
+ Misses 2762 2735 -27
🚀 New features to boost your workflow:
|
…tdown-timeout # Conflicts: # CHANGELOG.md
|
Thanks for taking this one, and sorry I am five days late to it. TLDR: this is the shape I asked for in #4359, and I left something out of that issue. What I measuredSame probe, same options ( The probe is one case added to your file, with a void SendRequest(std::shared_ptr<http_client::EventHandler> handler) noexcept override
{
held_ = std::move(handler); // never answers
}Your two cases pass in 0 ms, and I think they have to. They use the fake whose session answers inside The part I got wrong in the issueI pointed at const std::chrono::steady_clock::duration wait_interval = (std::min)(
std::chrono::duration_cast<std::chrono::steady_clock::duration>(options_.timeout),
timeout_steady);The Elasticsearch one always waits the full if (std::cv_status::no_timeout != synchronization_data_->force_flush_cv.wait_for(
lk_cv, std::chrono::seconds{options_.response_timeout_}))
{
break;
}With nothing to notify it there is exactly one 30 second wait, and that break skips the line below it that subtracts the elapsed time, so Giving the Elasticsearch The other half I would not change at all. Taking Unrelated, so you do not go looking: the red One last thing, on overlap. #4337 removes |
…ion directly ForceFlush() always waited response_timeout_ on its condition variable regardless of how much of the caller's own timeout was left, so Shutdown(timeout) could block far longer than requested whenever an export never completes. The wait is now clamped to whatever remains of the caller's deadline, matching the pattern already used by the OTLP HTTP client's ForceFlush(). Clamping surfaced a second, pre-existing bug: the loop treated any cv_status::no_timeout return from wait_for() as proof of completion, but the standard permits a spurious wakeup to report no_timeout too, indistinguishable from a real notification. A short-lived wait makes that far more likely to be hit, so ForceFlush() could report success without anything having actually finished. Completion is now always verified directly against finished_session_counter_ rather than inferred from the wait's return value. Reported by @thc1006 in review of open-telemetry#4523.
|
Thanks for the detailed repro, and no need to apologize for the timing. Added the clamp: Clamping it exposed a second bug your probe caught: the loop treated any Added Agreed on the rest, not touching anything around #4337/#4336, and thanks for flagging the msvc vcpkg failure as unrelated. cc: @thc1006 |
- clang-format wanted wait_interval's line-wrap reflowed. - IWYU wanted an explicit <algorithm> include for std::min. - HoldingSession held a shared_ptr back to the AsyncResponseHandler it was given, which itself holds a shared_ptr to the Session, forming a reference cycle that ASan/Valgrind flagged as a leak. Nothing on the AsyncResponseHandler destruction path touches finished_session_counter_ regardless of whether the handler is kept alive, so the retention was unnecessary; dropping the handler without storing it keeps the export "outstanding" just as well and removes the cycle.
|
Heads up on an interaction between this and #4502, with the fix below. My pull request is the one that breaks your test, so it seemed better to bring a patch than a report.
#4502 gives that destructor an outcome, because a handler torn down without one leaves a flush waiting forever. So the export becomes terminal before Measured on
Whichever of the two lands second turns the other red, so it is worth settling before either merges. The fix is for the fake to hold what its name says. The handler is parked in a variable the case owns rather than in the session, because @@ -180,30 +180,46 @@ class FakeHttpClient final : public http_client::HttpClient
class HoldingSession final : public http_client::Session
{
public:
+ explicit HoldingSession(std::shared_ptr<http_client::EventHandler> *parked) : parked_(parked) {}
+
std::shared_ptr<http_client::Request> CreateRequest() noexcept override
{
return std::make_shared<FakeRequest>();
}
- void SendRequest(std::shared_ptr<http_client::EventHandler>) noexcept override {}
+ // Parked where the case can see it, not in this session: the handler owns its session, so a
+ // session that owned the handler back would keep the pair alive.
+ void SendRequest(std::shared_ptr<http_client::EventHandler> handler) noexcept override
+ {
+ *parked_ = std::move(handler);
+ }
bool IsSessionActive() noexcept override { return true; }
bool CancelSession() noexcept override { return true; }
bool FinishSession() noexcept override { return true; }
+
+private:
+ std::shared_ptr<http_client::EventHandler> *parked_;
};
class HoldingHttpClient final : public http_client::HttpClient
{
public:
+ explicit HoldingHttpClient(std::shared_ptr<http_client::EventHandler> *parked) : parked_(parked)
+ {}
+
std::shared_ptr<http_client::Session> CreateSession(
opentelemetry::nostd::string_view) noexcept override
{
- return std::make_shared<HoldingSession>();
+ return std::make_shared<HoldingSession>(parked_);
}
bool CancelAllSessions() noexcept override { return true; }
bool FinishAllSessions() noexcept override { return true; }
void SetMaxSessionsPerConnection(std::size_t) noexcept override {}
+
+private:
+ std::shared_ptr<http_client::EventHandler> *parked_;
};
#endif // ENABLE_ASYNC_EXPORT
@@ -274,8 +290,10 @@ TEST(ElasticsearchLogsExporterTests, ShutdownReportsFlushCompletion)
#ifdef ENABLE_ASYNC_EXPORT
TEST(ElasticsearchLogsExporterTests, ShutdownClampsWaitToCallerTimeoutWhenExportIsOutstanding)
{
+ // Declared first so it outlives the client and the session that point at it.
+ std::shared_ptr<http_client::EventHandler> parked;
logs_exporter::ElasticsearchExporterOptions options;
- auto http_client = std::make_shared<HoldingHttpClient>();
+ auto http_client = std::make_shared<HoldingHttpClient>(&parked);
auto exporter = std::unique_ptr<sdklogs::LogRecordExporter>(
new logs_exporter::ElasticsearchLogRecordExporter(options, http_client));
@@ -291,6 +309,9 @@ TEST(ElasticsearchLogsExporterTests, ShutdownClampsWaitToCallerTimeoutWhenExport
EXPECT_FALSE(result);
EXPECT_LT(elapsed, std::chrono::seconds(1));
+
+ // Let the export finish now that the assertions are done.
+ parked.reset();
}
#endif // ENABLE_ASYNC_EXPORT
Built with |
…ENABLE_ASYNC_EXPORT
…lemetry#4502 Applied thc1006's patch from PR review: HoldingSession now parks the handler in a variable the test case owns instead of dropping it, so an export through it stays outstanding regardless of what open-telemetry#4502 changes about AsyncResponseHandler's destructor. A raw pointer to the parked variable (not a shared_ptr) keeps the session/client from forming a reference cycle with the handler.
|
Applied as-is, thanks for the thorough analysis and the ready patch. Pushed in Updated the comment above the two classes too, since the old one explained not retaining the handler as the way to avoid the cycle, which is no longer the mechanism now that we do retain it (a raw pointer to the parked variable, not a shared_ptr, is what avoids the cycle). cc: @thc1006 |
There was a problem hiding this comment.
The parked handler reads right, and the ForceFlush clamp plus the flush-then-cancel ordering in Shutdown are the changes I wanted to see. One thing left, and it is small.
The new shutdown re-check leaves its session with the client
CreateSession() has already run by the time the re-check fires, and the curl client keeps a shared_ptr to what it created:
std::lock_guard<std::mutex> lock_guard{sessions_m_};
sessions_.insert({session_id, session});
// FIXME: Session may leak if it does not call SendRequest
return session;That FIXME is upstream's own. Returning at the re-check happens before AsyncResponseHandler is constructed, so nothing on that path calls FinishSession(), and the local shared_ptr going out of scope cannot help while the client's map still holds one.
On main this window does not exist: Export()'s only shutdown check is above CreateSession(), and there is no other return between creating the session and SendRequest(). This pull request adds the first one.
#4502 does not cover it either. Its destructor fallback belongs to AsyncResponseHandler, which on this path is never built.
Measured on 9e30d1d6
A fake client that owns its sessions the way the real one does, gated so Shutdown() can land while the export is inside CreateSession(). Before the fix:
es_log_record_exporter_test.cc:416: Failure
0u <- client->Retained()
Which is: 1
the rejected session is still held by the client, so nothing will call FinishSession
es_log_record_exporter_test.cc:418: Failure
1u <- client->FinishCalls()
Which is: 0
handed back exactly once, not twice
[ FAILED ] ElasticsearchLogsExporterTests.ARejectedExportHandsItsSessionBack
The export correctly returns kFailure and sends nothing. It is the session that stays.
After the patch below: 7 of 7 pass with async on, 5 pass and 1 skips with it off, 30 repeats clean, and ASan with UBSan and LSan reports nothing.
The patch
Keeping the check and the counter under force_flush_m, and doing the cleanup after releasing it, so a FinishSession() that can reach a callback does not run inside the critical section:
@@ -442,19 +442,28 @@ sdk::common::ExportResult ElasticsearchLogRecordExporter::Export(
request->SetBody(body_vec);
#ifdef ENABLE_ASYNC_EXPORT
+ bool rejected = false;
// Send the request. Registration has to happen under the same lock Shutdown() takes to
// flip is_shutdown_ and snapshot session_counter_ (see ForceFlush()) - otherwise a session
// that passes the isShutdown() check above can still register after Shutdown() has already
// taken its snapshot, and ForceFlush() would return without ever having waited for it.
{
std::lock_guard<std::recursive_mutex> lock_guard{synchronization_data_->force_flush_m};
- if (isShutdown())
+ rejected = isShutdown();
+ if (!rejected)
{
- OTEL_INTERNAL_LOG_ERROR("[ES Log Exporter] Exporting "
- << records.size() << " log(s) failed, exporter is shutdown");
- return sdk::common::ExportResult::kFailure;
+ synchronization_data_->session_counter_.fetch_add(1, std::memory_order_release);
}
- synchronization_data_->session_counter_.fetch_add(1, std::memory_order_release);
+ }
+
+ // Outside the lock: the client owns the session until somebody hands it back, and this is the
+ // only path that can, since the handler that would do it later is never built.
+ if (rejected)
+ {
+ session->FinishSession();
+ OTEL_INTERNAL_LOG_ERROR("[ES Log Exporter] Exporting "
+ << records.size() << " log(s) failed, exporter is shutdown");
+ return sdk::common::ExportResult::kFailure;
}
std::size_t span_count = records.size();
auto synchronization_data = synchronization_data_;And the case. Two things are worth keeping in it. The fake client owns its sessions, because asserting only on the return value passes today: kFailure is already correct, it is the cleanup that is missing. And FinishSession() is counted rather than flagged, so handing a session back twice fails as loudly as not at all.
test diff
@@ -19,10 +19,14 @@
#include <gtest/gtest.h>
#include <array>
#include <chrono>
+#include <condition_variable>
#include <cstddef>
#include <cstdint>
+#include <mutex>
#include <string>
+#include <thread>
#include <utility>
+#include <vector>
#include "nlohmann/json.hpp"
namespace
@@ -106,6 +110,114 @@ class FakeSession final : public http_client::Session
bool FinishSession() noexcept override { return true; }
};
+// A client that owns what it creates, the way the curl client does: CreateSession() keeps a
+// reference in the client and only FinishSession() gives it back. CreateSession() also waits on
+// a gate, so a case can run Shutdown() while an export sits between its two shutdown checks.
+class RetainingSession final : public http_client::Session
+{
+public:
+ std::shared_ptr<http_client::Request> CreateRequest() noexcept override
+ {
+ return std::make_shared<FakeRequest>();
+ }
+
+ void SendRequest(std::shared_ptr<http_client::EventHandler>) noexcept override { sent_ = true; }
+
+ bool IsSessionActive() noexcept override { return finish_calls_ == 0; }
+ bool CancelSession() noexcept override { return true; }
+ // Counted rather than flagged: handing a session back twice is as wrong as not at all, and a
+ // flag reads the same either way.
+ bool FinishSession() noexcept override
+ {
+ ++finish_calls_;
+ return true;
+ }
+
+ bool sent_ = false;
+ std::size_t finish_calls_ = 0;
+};
+
+class RetainingHttpClient final : public http_client::HttpClient
+{
+public:
+ std::shared_ptr<http_client::Session> CreateSession(
+ opentelemetry::nostd::string_view) noexcept override
+ {
+ {
+ std::unique_lock<std::mutex> lock{gate_m_};
+ entered_ = true;
+ gate_cv_.notify_all();
+ gate_cv_.wait(lock, [this] { return released_; });
+ }
+ auto session = std::make_shared<RetainingSession>();
+ std::lock_guard<std::mutex> lock{sessions_m_};
+ sessions_.push_back(session);
+ return session;
+ }
+
+ bool CancelAllSessions() noexcept override { return true; }
+ bool FinishAllSessions() noexcept override { return true; }
+ void SetMaxSessionsPerConnection(std::size_t) noexcept override {}
+
+ void WaitUntilCreating()
+ {
+ std::unique_lock<std::mutex> lock{gate_m_};
+ gate_cv_.wait(lock, [this] { return entered_; });
+ }
+
+ void Release()
+ {
+ {
+ std::lock_guard<std::mutex> lock{gate_m_};
+ released_ = true;
+ }
+ gate_cv_.notify_all();
+ }
+
+ std::size_t Sent()
+ {
+ std::lock_guard<std::mutex> lock{sessions_m_};
+ std::size_t n = 0;
+ for (const auto &s : sessions_)
+ {
+ n += s->sent_ ? 1 : 0;
+ }
+ return n;
+ }
+
+ std::size_t FinishCalls()
+ {
+ std::lock_guard<std::mutex> lock{sessions_m_};
+ std::size_t n = 0;
+ for (const auto &s : sessions_)
+ {
+ n += s->finish_calls_;
+ }
+ return n;
+ }
+
+ // What the client is still holding that nobody handed back.
+ std::size_t Retained()
+ {
+ std::lock_guard<std::mutex> lock{sessions_m_};
+ std::size_t n = 0;
+ for (const auto &s : sessions_)
+ {
+ n += s->finish_calls_ == 0 ? 1 : 0;
+ }
+ return n;
+ }
+
+private:
+ std::mutex gate_m_;
+ std::condition_variable gate_cv_;
+ bool entered_ = false;
+ bool released_ = false;
+
+ std::mutex sessions_m_;
+ std::vector<std::shared_ptr<RetainingSession>> sessions_;
+};
+
class FakeHttpClient final : public http_client::HttpClient
{
public:
@@ -269,6 +381,44 @@ TEST(ElasticsearchLogsExporterTests, ShutdownClampsWaitToCallerTimeoutWhenExport
// Regression test: once Shutdown() has been called, any later Export() must fail rather than
// silently trying to register a session against an exporter that is already tearing down.
+// The case is only meaningful where the second shutdown check exists, but it is registered in
+// both builds: gtest_add_tests reads the source, so a case behind #ifdef is still handed to CTest
+// in the build that does not compile it and reports a pass it never ran.
+TEST(ElasticsearchLogsExporterTests, ARejectedExportHandsItsSessionBack)
+{
+#ifndef ENABLE_ASYNC_EXPORT
+ GTEST_SKIP() << "the shutdown re-check this covers is compiled only with async export";
+#else
+ auto client = std::make_shared<RetainingHttpClient>();
+ logs_exporter::ElasticsearchExporterOptions options;
+ auto exporter = std::unique_ptr<sdklogs::LogRecordExporter>(
+ new logs_exporter::ElasticsearchLogRecordExporter(options, client));
+
+ auto record = exporter->MakeRecordable();
+ record->SetBody("a record the exporter will refuse");
+ std::array<std::unique_ptr<sdklogs::Recordable>, 1> batch = {std::move(record)};
+
+ auto result = opentelemetry::sdk::common::ExportResult::kSuccess;
+ std::thread exporting([&] {
+ result = exporter->Export(
+ nostd::span<std::unique_ptr<sdklogs::Recordable>>(batch.data(), batch.size()));
+ });
+
+ // Shutdown lands while the export is inside CreateSession, so it sees no registered session
+ // and returns, and the export then meets the second check on its way back.
+ client->WaitUntilCreating();
+ exporter->Shutdown();
+ client->Release();
+ exporting.join();
+
+ EXPECT_EQ(opentelemetry::sdk::common::ExportResult::kFailure, result);
+ EXPECT_EQ(0u, client->Sent()) << "a rejected export must not send";
+ EXPECT_EQ(0u, client->Retained())
+ << "the rejected session is still held by the client, so nothing will call FinishSession";
+ EXPECT_EQ(1u, client->FinishCalls()) << "handed back exactly once, not twice";
+#endif
+}
+
TEST(ElasticsearchLogsExporterTests, ExportAfterShutdownFails)
{
logs_exporter::ElasticsearchExporterOptions options;One note on the guard. I wrote the case so it is compiled in both builds and skips at run time, rather than sitting behind #ifdef. gtest_add_tests reads the source, so a case behind #ifdef is still registered with CTest in the build that does not compile it, and reports a pass it never ran. ShutdownClampsWaitToCallerTimeoutWhenExportIsOutstanding has that shape today, which is worth a separate look but not something to fix here.
Nothing else from me. This does not need the #4337 refactor. One thing in the other direction, since I own that one: #4337 still has the old Shutdown that ignores its timeout and returns true unconditionally, so whichever of us rebases second has to keep the ordering this pull request establishes rather than the one mine still carries. I have noted that on #4337 so it does not get lost.
…own() The second shutdown check added earlier runs after CreateSession(), so a real client that retains a shared_ptr to every session it creates (e.g. curl) never gets that session back on the rejected path, since the handler that would normally call FinishSession() is never built. Move the rejection decision under the lock but call session->FinishSession() outside it before returning failure, so the client's own session bookkeeping is not left dangling. Applied thc1006's patch and regression test (ARejectedExportHandsItsSessionBack) from PR review, using a RetainingHttpClient/RetainingSession pair that mimics curl's session retention and gates CreateSession() so Shutdown() can land mid-export. Verified locally: 7/7 pass with async export on, 5 pass + 1 skip with it off, --gtest_repeat=30 on the new test is stable, and ASan+UBSan+LSan report nothing.
|
The rejection is now decided under the lock but Verified locally: 7/7 pass with async export on, 5 pass + 1 skip with it off, Noted the point on |
This branch has not been deployed
2 participants
Fixes #4359.
ElasticsearchLogRecordExporter::Shutdown()never read itstimeoutparameter and always returnedtrue, whether or not anything had actually flushed.Changes
Shutdown()now flushes pending exports against the caller's deadline viaForceFlush(timeout)before cancelling sessions, and returns what that flush reported: mirroring the patternOtlpHttpClient::Shutdown/OtlpGrpcClient::Shutdownalready use. Cancelling happens after the flush, not before, so there's still something to wait for.Export()could pass itsisShutdown()check and still register a session (bumpsession_counter_) afterShutdown()had already taken its snapshot of that counter insideForceFlush(), so the flush could return success without ever having waited for that session. Both theis_shutdown_flag and session registration now go through the sameforce_flush_mlock that already guardedForceFlush().Testing
Added two tests against the existing fake
HttpClient/Session/Request/Responsetest doubles:ShutdownReportsFlushCompletion: export completes, thenShutdown(timeout)returnstrue.ExportAfterShutdownFails:Export()afterShutdown()returnskFailure.Built and ran the exporter's test suite under both configurations (
OTELCPP_WITH_ASYNC_EXPORT_PREVIEWon and off), since the registration race only exists on the async path: both pass.