Skip to content

Commit 2b467f6

Browse files
committed
[TEST] Hand the deferred handler over through the promise, and guard the log assertions
Three things the cases were getting away with. The deferred helper published the handler into a slot beside the promise and signalled with the promise. EXPECT_EQ is not fatal, so when the five second handoff wait timed out the helper carried on and read that slot while the exporter's thread could still be writing it. A wait that times out has not observed the promise becoming ready, so those two accesses have nothing ordering them and the failure path was itself undefined. The handler now travels in the promise and is taken from the future, and each precondition reports and returns instead of leaving the rest of the helper to run on state it just said was wrong. The cases that read the exporter's error lines now sit on their own fixture. Below error level OTEL_INTERNAL_LOG_ERROR expands to nothing rather than being filtered, so with -DOTEL_INTERNAL_LOG_LEVEL=0 the ones expecting the winner to report failed on correct code and the ones expecting silence from the loser passed without testing anything. Measured: with the fixture, twelve pass and those six skip; move one back and it fails expecting one line and finding none. The accepted bulk body now names an index. open-telemetry#4297 requires a string _index in every index acknowledgement, so without it every success case here would be read as a failure once these two meet, which is the opposite of what this file's comment promised. The two deferred cases are renamed for what they hold. The session publishes the handler before SendRequest() returns, so they cover a callback delivered after the handoff while the export is still running, not a waiter proven to be inside cv_.wait(). Signed-off-by: thc1006 <84045975+thc1006@users.noreply.github.com>
1 parent d7f02bb commit 2b467f6

1 file changed

Lines changed: 72 additions & 31 deletions

File tree

‎exporters/elasticsearch/test/es_log_record_exporter_test.cc‎

Lines changed: 72 additions & 31 deletions
Original file line numberDiff line numberDiff line change
@@ -284,7 +284,7 @@ namespace http_client = opentelemetry::ext::http::client;
284284
// acknowledged operation result carrying a 2xx status, so these cases keep meaning the
285285
// same thing whichever success check is in place.
286286
constexpr const char *kAcceptedBody =
287-
R"({"took":30,"errors":false,"items":[{"index":{"status":201,"_shards":{"failed" : 0}}}]})";
287+
R"({"took":30,"errors":false,"items":[{"index":{"_index":"logs","status":201,"_shards":{"failed" : 0}}}]})";
288288

289289
class FakeResponse : public http_client::Response
290290
{
@@ -353,46 +353,46 @@ class FakeSession : public http_client::Session
353353
class DeferredSession : public http_client::Session
354354
{
355355
public:
356-
DeferredSession(std::shared_ptr<http_client::EventHandler> *slot, std::promise<void> *arrived)
357-
: slot_(slot), arrived_(arrived)
356+
explicit DeferredSession(std::promise<std::shared_ptr<http_client::EventHandler>> *arrived)
357+
: arrived_(arrived)
358358
{}
359359

360360
std::shared_ptr<http_client::Request> CreateRequest() noexcept override
361361
{
362362
return std::make_shared<FakeRequest>();
363363
}
364+
// The handler travels in the promise rather than beside it. A waiter that times out has not
365+
// observed the promise becoming ready and so is not synchronized with this thread, which would
366+
// make a handler read on that path a race with this write.
364367
void SendRequest(std::shared_ptr<http_client::EventHandler> handler) noexcept override
365368
{
366-
*slot_ = std::move(handler);
367-
arrived_->set_value();
369+
arrived_->set_value(std::move(handler));
368370
}
369371
bool IsSessionActive() noexcept override { return true; }
370372
bool CancelSession() noexcept override { return true; }
371373
bool FinishSession() noexcept override { return true; }
372374

373375
private:
374-
std::shared_ptr<http_client::EventHandler> *slot_;
375-
std::promise<void> *arrived_;
376+
std::promise<std::shared_ptr<http_client::EventHandler>> *arrived_;
376377
};
377378

378379
class DeferredHttpClient : public http_client::HttpClient
379380
{
380381
public:
381-
DeferredHttpClient(std::shared_ptr<http_client::EventHandler> *slot, std::promise<void> *arrived)
382-
: slot_(slot), arrived_(arrived)
382+
explicit DeferredHttpClient(std::promise<std::shared_ptr<http_client::EventHandler>> *arrived)
383+
: arrived_(arrived)
383384
{}
384385

385386
std::shared_ptr<http_client::Session> CreateSession(nostd::string_view) noexcept override
386387
{
387-
return std::make_shared<DeferredSession>(slot_, arrived_);
388+
return std::make_shared<DeferredSession>(arrived_);
388389
}
389390
bool CancelAllSessions() noexcept override { return true; }
390391
bool FinishAllSessions() noexcept override { return true; }
391392
void SetMaxSessionsPerConnection(std::size_t) noexcept override {}
392393

393394
private:
394-
std::shared_ptr<http_client::EventHandler> *slot_;
395-
std::promise<void> *arrived_;
395+
std::promise<std::shared_ptr<http_client::EventHandler>> *arrived_;
396396
};
397397

398398
class FakeHttpClient : public http_client::HttpClient
@@ -438,6 +438,25 @@ class ElasticsearchLogsExporterSyncTests : public ::testing::Test
438438
#endif
439439
}
440440
};
441+
442+
// For the cases that read the lines the exporter writes rather than only the result it returns.
443+
// Below error level OTEL_INTERNAL_LOG_ERROR expands to nothing, so the line is not filtered at
444+
// runtime, it does not exist: a case expecting the winner to report one would fail on correct
445+
// code, and one expecting silence from the loser would pass without testing anything.
446+
class ElasticsearchLogsExporterSyncLoggingTests : public ::testing::Test
447+
{
448+
protected:
449+
void SetUp() override
450+
{
451+
#ifdef ENABLE_ASYNC_EXPORT
452+
GTEST_SKIP() << "Export() returns without waiting when async export is enabled";
453+
#elif OTEL_INTERNAL_LOG_LEVEL < OTEL_INTERNAL_LOG_LEVEL_ERROR
454+
// One skip point, because GTEST_SKIP returns and a second one after it would leave the rest of
455+
// this body unreachable, which MSVC reports as C4702 under maintainer mode.
456+
GTEST_SKIP() << "the exporter's error lines are compiled out below error level";
457+
#endif
458+
}
459+
};
441460
} // namespace
442461

443462
namespace
@@ -451,15 +470,18 @@ struct ParkedExport
451470
std::shared_ptr<DeferredHttpClient> client;
452471
std::shared_ptr<logs_exporter::ElasticsearchLogRecordExporter> exporter;
453472
std::shared_ptr<http_client::EventHandler> handler;
454-
std::promise<void> arrived;
473+
std::promise<std::shared_ptr<http_client::EventHandler>> arrived;
455474
std::promise<opentelemetry::sdk::common::ExportResult> done;
456475
std::future<opentelemetry::sdk::common::ExportResult> finished;
457476
};
458477

478+
// Answers with nullptr rather than a half prepared handle when a precondition does not hold. Each
479+
// step below is something a case needs before it can mean anything, and carrying on past one of
480+
// them reads state the other thread is still writing.
459481
std::shared_ptr<ParkedExport> StartParkedExport()
460482
{
461483
auto parked = std::make_shared<ParkedExport>();
462-
parked->client = std::make_shared<DeferredHttpClient>(&parked->handler, &parked->arrived);
484+
parked->client = std::make_shared<DeferredHttpClient>(&parked->arrived);
463485

464486
logs_exporter::ElasticsearchExporterOptions options;
465487
parked->exporter =
@@ -474,23 +496,42 @@ std::shared_ptr<ParkedExport> StartParkedExport()
474496
parked->exporter->Export(nostd::span<std::unique_ptr<sdklogs::Recordable>>(&record, 1)));
475497
}).detach();
476498

477-
EXPECT_EQ(std::future_status::ready, reached.wait_for(std::chrono::seconds{5}))
478-
<< "the exporter never sent the request";
479-
EXPECT_TRUE(parked->handler) << "the session was not given a handler";
480-
EXPECT_EQ(std::future_status::timeout, finished.wait_for(std::chrono::milliseconds{100}))
481-
<< "the export returned without waiting for anything";
499+
if (std::future_status::ready != reached.wait_for(std::chrono::seconds{5}))
500+
{
501+
ADD_FAILURE() << "the exporter never handed off the request handler";
502+
return nullptr;
503+
}
504+
505+
parked->handler = reached.get();
506+
if (!parked->handler)
507+
{
508+
ADD_FAILURE() << "the session was handed a null handler";
509+
return nullptr;
510+
}
511+
512+
if (std::future_status::timeout != finished.wait_for(std::chrono::milliseconds{100}))
513+
{
514+
ADD_FAILURE() << "the export returned before any callback was delivered";
515+
return nullptr;
516+
}
482517

483518
parked->finished = std::move(finished);
484519
return parked;
485520
}
486521
} // namespace
487522

488-
// A terminal error delivered once the waiter is parked has to end the wait, which is the half the
489-
// notification is responsible for. Without these, removing cv_.notify_all() keeps this file green.
490-
TEST_F(ElasticsearchLogsExporterSyncTests, AReadErrorAfterTheWaiterParksWakesTheExport)
523+
// A terminal error that arrives after the handler has been handed off, while the export is still
524+
// running, has to end the wait, which is the half the notification is responsible for. Without
525+
// these two, removing cv_.notify_all() keeps this file green.
526+
//
527+
// The handoff is what they hold, not the parking: the session publishes the handler before
528+
// SendRequest() returns, so the export need not have reached cv_.wait() when the callback is
529+
// delivered. Holding that would want a wait entry seam in production code, which is not worth the
530+
// API it would add.
531+
TEST_F(ElasticsearchLogsExporterSyncTests, AReadErrorAfterTheHandoffEndsTheExport)
491532
{
492533
auto parked = StartParkedExport();
493-
ASSERT_TRUE(parked->handler);
534+
ASSERT_NE(nullptr, parked);
494535

495536
parked->handler->OnEvent(http_client::SessionState::ReadError, "");
496537

@@ -501,10 +542,10 @@ TEST_F(ElasticsearchLogsExporterSyncTests, AReadErrorAfterTheWaiterParksWakesThe
501542

502543
// The same for the success half, which also holds that the wait is a wait: a waitForResponse that
503544
// only read the current state would answer before this response arrives.
504-
TEST_F(ElasticsearchLogsExporterSyncTests, AResponseAfterTheWaiterParksWakesTheExport)
545+
TEST_F(ElasticsearchLogsExporterSyncTests, AResponseAfterTheHandoffEndsTheExport)
505546
{
506547
auto parked = StartParkedExport();
507-
ASSERT_TRUE(parked->handler);
548+
ASSERT_NE(nullptr, parked);
508549

509550
FakeResponse response(200, kAcceptedBody);
510551
parked->handler->OnResponse(response);
@@ -514,7 +555,7 @@ TEST_F(ElasticsearchLogsExporterSyncTests, AResponseAfterTheWaiterParksWakesTheE
514555
EXPECT_EQ(opentelemetry::sdk::common::ExportResult::kSuccess, parked->finished.get());
515556
}
516557

517-
TEST_F(ElasticsearchLogsExporterSyncTests, ResponseRecordedBeforeTheWaitIsStillSeen)
558+
TEST_F(ElasticsearchLogsExporterSyncLoggingTests, ResponseRecordedBeforeTheWaitIsStillSeen)
518559
{
519560
const auto result = ExportWith([](http_client::EventHandler &handler) {
520561
FakeResponse response(200, kAcceptedBody);
@@ -596,7 +637,7 @@ std::size_t CountExporterErrors(const std::vector<std::string> &lines)
596637
// The winning side of the same rule. Without this, a recordCompletion that still records the
597638
// outcome but always answers "you did not decide it" passes every other case here while the
598639
// exporter goes silent about every failure it reports to its caller.
599-
TEST_F(ElasticsearchLogsExporterSyncTests, AWinningTerminalErrorIsReportedExactlyOnce)
640+
TEST_F(ElasticsearchLogsExporterSyncLoggingTests, AWinningTerminalErrorIsReportedExactlyOnce)
600641
{
601642
for (const auto &terminal : kErrorTerminalStates)
602643
{
@@ -630,7 +671,7 @@ TEST_F(ElasticsearchLogsExporterSyncTests, AWinningTerminalErrorIsReportedExactl
630671

631672
// The shared curl client has reported more than one terminal event for one request, see #4360,
632673
// so which of two failures is described matters rather than only how many arrive.
633-
TEST_F(ElasticsearchLogsExporterSyncTests, OnlyTheFirstTerminalFailureIsReported)
674+
TEST_F(ElasticsearchLogsExporterSyncLoggingTests, OnlyTheFirstTerminalFailureIsReported)
634675
{
635676
auto capturing = nostd::shared_ptr<internal_log::LogHandler>(new ErrorCapturingLogHandler());
636677
const auto previous = internal_log::GlobalLogHandler::GetLogHandler();
@@ -664,7 +705,7 @@ TEST_F(ElasticsearchLogsExporterSyncTests, OnlyTheFirstTerminalFailureIsReported
664705
// covers a completion recorded before the waiter arrives but never two callbacks at once. This
665706
// releases a response and a read error together. Either may win, and the point is that the result
666707
// and the diagnostic never disagree about which one did.
667-
TEST_F(ElasticsearchLogsExporterSyncTests, AResponseRacingAReadErrorAgree)
708+
TEST_F(ElasticsearchLogsExporterSyncLoggingTests, AResponseRacingAReadErrorAgree)
668709
{
669710
for (int attempt = 0; attempt < 50; ++attempt)
670711
{
@@ -726,7 +767,7 @@ TEST_F(ElasticsearchLogsExporterSyncTests, AResponseRacingAReadErrorAgree)
726767
}
727768

728769
// The losing side, over the same table rather than three of the nine.
729-
TEST_F(ElasticsearchLogsExporterSyncTests, NoTerminalErrorAfterAResponseIsReported)
770+
TEST_F(ElasticsearchLogsExporterSyncLoggingTests, NoTerminalErrorAfterAResponseIsReported)
730771
{
731772
for (const auto &terminal : kErrorTerminalStates)
732773
{
@@ -751,7 +792,7 @@ TEST_F(ElasticsearchLogsExporterSyncTests, NoTerminalErrorAfterAResponseIsReport
751792
}
752793
}
753794

754-
TEST_F(ElasticsearchLogsExporterSyncTests, ALateTerminalFailureDoesNotClaimTheExportFailed)
795+
TEST_F(ElasticsearchLogsExporterSyncLoggingTests, ALateTerminalFailureDoesNotClaimTheExportFailed)
755796
{
756797
auto capturing = nostd::shared_ptr<internal_log::LogHandler>(new ErrorCapturingLogHandler());
757798
const auto previous = internal_log::GlobalLogHandler::GetLogHandler();

0 commit comments

Comments
 (0)