Skip to content

Commit 9a14aa8

Browse files
committed
Fix session leak when Export() is rejected mid-CreateSession by Shutdown()
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.
1 parent a2372e9 commit 9a14aa8

2 files changed

Lines changed: 164 additions & 5 deletions

File tree

‎exporters/elasticsearch/src/es_log_record_exporter.cc‎

Lines changed: 14 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -442,19 +442,28 @@ sdk::common::ExportResult ElasticsearchLogRecordExporter::Export(
442442
request->SetBody(body_vec);
443443

444444
#ifdef ENABLE_ASYNC_EXPORT
445+
bool rejected = false;
445446
// Send the request. Registration has to happen under the same lock Shutdown() takes to
446447
// flip is_shutdown_ and snapshot session_counter_ (see ForceFlush()) - otherwise a session
447448
// that passes the isShutdown() check above can still register after Shutdown() has already
448449
// taken its snapshot, and ForceFlush() would return without ever having waited for it.
449450
{
450451
std::lock_guard<std::recursive_mutex> lock_guard{synchronization_data_->force_flush_m};
451-
if (isShutdown())
452+
rejected = isShutdown();
453+
if (!rejected)
452454
{
453-
OTEL_INTERNAL_LOG_ERROR("[ES Log Exporter] Exporting "
454-
<< records.size() << " log(s) failed, exporter is shutdown");
455-
return sdk::common::ExportResult::kFailure;
455+
synchronization_data_->session_counter_.fetch_add(1, std::memory_order_release);
456456
}
457-
synchronization_data_->session_counter_.fetch_add(1, std::memory_order_release);
457+
}
458+
459+
// Outside the lock: the client owns the session until somebody hands it back, and this is the
460+
// only path that can, since the handler that would do it later is never built.
461+
if (rejected)
462+
{
463+
session->FinishSession();
464+
OTEL_INTERNAL_LOG_ERROR("[ES Log Exporter] Exporting "
465+
<< records.size() << " log(s) failed, exporter is shutdown");
466+
return sdk::common::ExportResult::kFailure;
458467
}
459468
std::size_t span_count = records.size();
460469
auto synchronization_data = synchronization_data_;

‎exporters/elasticsearch/test/es_log_record_exporter_test.cc‎

Lines changed: 150 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -19,10 +19,14 @@
1919
#include <gtest/gtest.h>
2020
#include <array>
2121
#include <chrono>
22+
#include <condition_variable>
2223
#include <cstddef>
2324
#include <cstdint>
25+
#include <mutex>
2426
#include <string>
27+
#include <thread>
2528
#include <utility>
29+
#include <vector>
2630
#include "nlohmann/json.hpp"
2731

2832
namespace
@@ -106,6 +110,114 @@ class FakeSession final : public http_client::Session
106110
bool FinishSession() noexcept override { return true; }
107111
};
108112

113+
// A client that owns what it creates, the way the curl client does: CreateSession() keeps a
114+
// reference in the client and only FinishSession() gives it back. CreateSession() also waits on
115+
// a gate, so a case can run Shutdown() while an export sits between its two shutdown checks.
116+
class RetainingSession final : public http_client::Session
117+
{
118+
public:
119+
std::shared_ptr<http_client::Request> CreateRequest() noexcept override
120+
{
121+
return std::make_shared<FakeRequest>();
122+
}
123+
124+
void SendRequest(std::shared_ptr<http_client::EventHandler>) noexcept override { sent_ = true; }
125+
126+
bool IsSessionActive() noexcept override { return finish_calls_ == 0; }
127+
bool CancelSession() noexcept override { return true; }
128+
// Counted rather than flagged: handing a session back twice is as wrong as not at all, and a
129+
// flag reads the same either way.
130+
bool FinishSession() noexcept override
131+
{
132+
++finish_calls_;
133+
return true;
134+
}
135+
136+
bool sent_ = false;
137+
std::size_t finish_calls_ = 0;
138+
};
139+
140+
class RetainingHttpClient final : public http_client::HttpClient
141+
{
142+
public:
143+
std::shared_ptr<http_client::Session> CreateSession(
144+
opentelemetry::nostd::string_view) noexcept override
145+
{
146+
{
147+
std::unique_lock<std::mutex> lock{gate_m_};
148+
entered_ = true;
149+
gate_cv_.notify_all();
150+
gate_cv_.wait(lock, [this] { return released_; });
151+
}
152+
auto session = std::make_shared<RetainingSession>();
153+
std::lock_guard<std::mutex> lock{sessions_m_};
154+
sessions_.push_back(session);
155+
return session;
156+
}
157+
158+
bool CancelAllSessions() noexcept override { return true; }
159+
bool FinishAllSessions() noexcept override { return true; }
160+
void SetMaxSessionsPerConnection(std::size_t) noexcept override {}
161+
162+
void WaitUntilCreating()
163+
{
164+
std::unique_lock<std::mutex> lock{gate_m_};
165+
gate_cv_.wait(lock, [this] { return entered_; });
166+
}
167+
168+
void Release()
169+
{
170+
{
171+
std::lock_guard<std::mutex> lock{gate_m_};
172+
released_ = true;
173+
}
174+
gate_cv_.notify_all();
175+
}
176+
177+
std::size_t Sent()
178+
{
179+
std::lock_guard<std::mutex> lock{sessions_m_};
180+
std::size_t n = 0;
181+
for (const auto &s : sessions_)
182+
{
183+
n += s->sent_ ? 1 : 0;
184+
}
185+
return n;
186+
}
187+
188+
std::size_t FinishCalls()
189+
{
190+
std::lock_guard<std::mutex> lock{sessions_m_};
191+
std::size_t n = 0;
192+
for (const auto &s : sessions_)
193+
{
194+
n += s->finish_calls_;
195+
}
196+
return n;
197+
}
198+
199+
// What the client is still holding that nobody handed back.
200+
std::size_t Retained()
201+
{
202+
std::lock_guard<std::mutex> lock{sessions_m_};
203+
std::size_t n = 0;
204+
for (const auto &s : sessions_)
205+
{
206+
n += s->finish_calls_ == 0 ? 1 : 0;
207+
}
208+
return n;
209+
}
210+
211+
private:
212+
std::mutex gate_m_;
213+
std::condition_variable gate_cv_;
214+
bool entered_ = false;
215+
bool released_ = false;
216+
217+
std::mutex sessions_m_;
218+
std::vector<std::shared_ptr<RetainingSession>> sessions_;
219+
};
220+
109221
class FakeHttpClient final : public http_client::HttpClient
110222
{
111223
public:
@@ -269,6 +381,44 @@ TEST(ElasticsearchLogsExporterTests, ShutdownClampsWaitToCallerTimeoutWhenExport
269381

270382
// Regression test: once Shutdown() has been called, any later Export() must fail rather than
271383
// silently trying to register a session against an exporter that is already tearing down.
384+
// The case is only meaningful where the second shutdown check exists, but it is registered in
385+
// both builds: gtest_add_tests reads the source, so a case behind #ifdef is still handed to CTest
386+
// in the build that does not compile it and reports a pass it never ran.
387+
TEST(ElasticsearchLogsExporterTests, ARejectedExportHandsItsSessionBack)
388+
{
389+
#ifndef ENABLE_ASYNC_EXPORT
390+
GTEST_SKIP() << "the shutdown re-check this covers is compiled only with async export";
391+
#else
392+
auto client = std::make_shared<RetainingHttpClient>();
393+
logs_exporter::ElasticsearchExporterOptions options;
394+
auto exporter = std::unique_ptr<sdklogs::LogRecordExporter>(
395+
new logs_exporter::ElasticsearchLogRecordExporter(options, client));
396+
397+
auto record = exporter->MakeRecordable();
398+
record->SetBody("a record the exporter will refuse");
399+
std::array<std::unique_ptr<sdklogs::Recordable>, 1> batch = {std::move(record)};
400+
401+
auto result = opentelemetry::sdk::common::ExportResult::kSuccess;
402+
std::thread exporting([&] {
403+
result = exporter->Export(
404+
nostd::span<std::unique_ptr<sdklogs::Recordable>>(batch.data(), batch.size()));
405+
});
406+
407+
// Shutdown lands while the export is inside CreateSession, so it sees no registered session
408+
// and returns, and the export then meets the second check on its way back.
409+
client->WaitUntilCreating();
410+
exporter->Shutdown();
411+
client->Release();
412+
exporting.join();
413+
414+
EXPECT_EQ(opentelemetry::sdk::common::ExportResult::kFailure, result);
415+
EXPECT_EQ(0u, client->Sent()) << "a rejected export must not send";
416+
EXPECT_EQ(0u, client->Retained())
417+
<< "the rejected session is still held by the client, so nothing will call FinishSession";
418+
EXPECT_EQ(1u, client->FinishCalls()) << "handed back exactly once, not twice";
419+
#endif
420+
}
421+
272422
TEST(ElasticsearchLogsExporterTests, ExportAfterShutdownFails)
273423
{
274424
logs_exporter::ElasticsearchExporterOptions options;

0 commit comments

Comments
 (0)