Skip to content

Commit a85e705

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 f1ec07c commit a85e705

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
@@ -165,7 +165,7 @@ namespace http_client = opentelemetry::ext::http::client;
165165
// acknowledged operation result carrying a 2xx status, so these cases keep meaning the
166166
// same thing whichever success check is in place.
167167
constexpr const char *kAcceptedBody =
168-
R"({"took":30,"errors":false,"items":[{"index":{"status":201,"_shards":{"failed" : 0}}}]})";
168+
R"({"took":30,"errors":false,"items":[{"index":{"_index":"logs","status":201,"_shards":{"failed" : 0}}}]})";
169169

170170
class FakeResponse : public http_client::Response
171171
{
@@ -234,46 +234,46 @@ class FakeSession : public http_client::Session
234234
class DeferredSession : public http_client::Session
235235
{
236236
public:
237-
DeferredSession(std::shared_ptr<http_client::EventHandler> *slot, std::promise<void> *arrived)
238-
: slot_(slot), arrived_(arrived)
237+
explicit DeferredSession(std::promise<std::shared_ptr<http_client::EventHandler>> *arrived)
238+
: arrived_(arrived)
239239
{}
240240

241241
std::shared_ptr<http_client::Request> CreateRequest() noexcept override
242242
{
243243
return std::make_shared<FakeRequest>();
244244
}
245+
// The handler travels in the promise rather than beside it. A waiter that times out has not
246+
// observed the promise becoming ready and so is not synchronized with this thread, which would
247+
// make a handler read on that path a race with this write.
245248
void SendRequest(std::shared_ptr<http_client::EventHandler> handler) noexcept override
246249
{
247-
*slot_ = std::move(handler);
248-
arrived_->set_value();
250+
arrived_->set_value(std::move(handler));
249251
}
250252
bool IsSessionActive() noexcept override { return true; }
251253
bool CancelSession() noexcept override { return true; }
252254
bool FinishSession() noexcept override { return true; }
253255

254256
private:
255-
std::shared_ptr<http_client::EventHandler> *slot_;
256-
std::promise<void> *arrived_;
257+
std::promise<std::shared_ptr<http_client::EventHandler>> *arrived_;
257258
};
258259

259260
class DeferredHttpClient : public http_client::HttpClient
260261
{
261262
public:
262-
DeferredHttpClient(std::shared_ptr<http_client::EventHandler> *slot, std::promise<void> *arrived)
263-
: slot_(slot), arrived_(arrived)
263+
explicit DeferredHttpClient(std::promise<std::shared_ptr<http_client::EventHandler>> *arrived)
264+
: arrived_(arrived)
264265
{}
265266

266267
std::shared_ptr<http_client::Session> CreateSession(nostd::string_view) noexcept override
267268
{
268-
return std::make_shared<DeferredSession>(slot_, arrived_);
269+
return std::make_shared<DeferredSession>(arrived_);
269270
}
270271
bool CancelAllSessions() noexcept override { return true; }
271272
bool FinishAllSessions() noexcept override { return true; }
272273
void SetMaxSessionsPerConnection(std::size_t) noexcept override {}
273274

274275
private:
275-
std::shared_ptr<http_client::EventHandler> *slot_;
276-
std::promise<void> *arrived_;
276+
std::promise<std::shared_ptr<http_client::EventHandler>> *arrived_;
277277
};
278278

279279
class FakeHttpClient : public http_client::HttpClient
@@ -319,6 +319,25 @@ class ElasticsearchLogsExporterSyncTests : public ::testing::Test
319319
#endif
320320
}
321321
};
322+
323+
// For the cases that read the lines the exporter writes rather than only the result it returns.
324+
// Below error level OTEL_INTERNAL_LOG_ERROR expands to nothing, so the line is not filtered at
325+
// runtime, it does not exist: a case expecting the winner to report one would fail on correct
326+
// code, and one expecting silence from the loser would pass without testing anything.
327+
class ElasticsearchLogsExporterSyncLoggingTests : public ::testing::Test
328+
{
329+
protected:
330+
void SetUp() override
331+
{
332+
#ifdef ENABLE_ASYNC_EXPORT
333+
GTEST_SKIP() << "Export() returns without waiting when async export is enabled";
334+
#elif OTEL_INTERNAL_LOG_LEVEL < OTEL_INTERNAL_LOG_LEVEL_ERROR
335+
// One skip point, because GTEST_SKIP returns and a second one after it would leave the rest of
336+
// this body unreachable, which MSVC reports as C4702 under maintainer mode.
337+
GTEST_SKIP() << "the exporter's error lines are compiled out below error level";
338+
#endif
339+
}
340+
};
322341
} // namespace
323342

324343
namespace
@@ -332,15 +351,18 @@ struct ParkedExport
332351
std::shared_ptr<DeferredHttpClient> client;
333352
std::shared_ptr<logs_exporter::ElasticsearchLogRecordExporter> exporter;
334353
std::shared_ptr<http_client::EventHandler> handler;
335-
std::promise<void> arrived;
354+
std::promise<std::shared_ptr<http_client::EventHandler>> arrived;
336355
std::promise<opentelemetry::sdk::common::ExportResult> done;
337356
std::future<opentelemetry::sdk::common::ExportResult> finished;
338357
};
339358

359+
// Answers with nullptr rather than a half prepared handle when a precondition does not hold. Each
360+
// step below is something a case needs before it can mean anything, and carrying on past one of
361+
// them reads state the other thread is still writing.
340362
std::shared_ptr<ParkedExport> StartParkedExport()
341363
{
342364
auto parked = std::make_shared<ParkedExport>();
343-
parked->client = std::make_shared<DeferredHttpClient>(&parked->handler, &parked->arrived);
365+
parked->client = std::make_shared<DeferredHttpClient>(&parked->arrived);
344366

345367
logs_exporter::ElasticsearchExporterOptions options;
346368
parked->exporter =
@@ -355,23 +377,42 @@ std::shared_ptr<ParkedExport> StartParkedExport()
355377
parked->exporter->Export(nostd::span<std::unique_ptr<sdklogs::Recordable>>(&record, 1)));
356378
}).detach();
357379

358-
EXPECT_EQ(std::future_status::ready, reached.wait_for(std::chrono::seconds{5}))
359-
<< "the exporter never sent the request";
360-
EXPECT_TRUE(parked->handler) << "the session was not given a handler";
361-
EXPECT_EQ(std::future_status::timeout, finished.wait_for(std::chrono::milliseconds{100}))
362-
<< "the export returned without waiting for anything";
380+
if (std::future_status::ready != reached.wait_for(std::chrono::seconds{5}))
381+
{
382+
ADD_FAILURE() << "the exporter never handed off the request handler";
383+
return nullptr;
384+
}
385+
386+
parked->handler = reached.get();
387+
if (!parked->handler)
388+
{
389+
ADD_FAILURE() << "the session was handed a null handler";
390+
return nullptr;
391+
}
392+
393+
if (std::future_status::timeout != finished.wait_for(std::chrono::milliseconds{100}))
394+
{
395+
ADD_FAILURE() << "the export returned before any callback was delivered";
396+
return nullptr;
397+
}
363398

364399
parked->finished = std::move(finished);
365400
return parked;
366401
}
367402
} // namespace
368403

369-
// A terminal error delivered once the waiter is parked has to end the wait, which is the half the
370-
// notification is responsible for. Without these, removing cv_.notify_all() keeps this file green.
371-
TEST_F(ElasticsearchLogsExporterSyncTests, AReadErrorAfterTheWaiterParksWakesTheExport)
404+
// A terminal error that arrives after the handler has been handed off, while the export is still
405+
// running, has to end the wait, which is the half the notification is responsible for. Without
406+
// these two, removing cv_.notify_all() keeps this file green.
407+
//
408+
// The handoff is what they hold, not the parking: the session publishes the handler before
409+
// SendRequest() returns, so the export need not have reached cv_.wait() when the callback is
410+
// delivered. Holding that would want a wait entry seam in production code, which is not worth the
411+
// API it would add.
412+
TEST_F(ElasticsearchLogsExporterSyncTests, AReadErrorAfterTheHandoffEndsTheExport)
372413
{
373414
auto parked = StartParkedExport();
374-
ASSERT_TRUE(parked->handler);
415+
ASSERT_NE(nullptr, parked);
375416

376417
parked->handler->OnEvent(http_client::SessionState::ReadError, "");
377418

@@ -382,10 +423,10 @@ TEST_F(ElasticsearchLogsExporterSyncTests, AReadErrorAfterTheWaiterParksWakesThe
382423

383424
// The same for the success half, which also holds that the wait is a wait: a waitForResponse that
384425
// only read the current state would answer before this response arrives.
385-
TEST_F(ElasticsearchLogsExporterSyncTests, AResponseAfterTheWaiterParksWakesTheExport)
426+
TEST_F(ElasticsearchLogsExporterSyncTests, AResponseAfterTheHandoffEndsTheExport)
386427
{
387428
auto parked = StartParkedExport();
388-
ASSERT_TRUE(parked->handler);
429+
ASSERT_NE(nullptr, parked);
389430

390431
FakeResponse response(200, kAcceptedBody);
391432
parked->handler->OnResponse(response);
@@ -395,7 +436,7 @@ TEST_F(ElasticsearchLogsExporterSyncTests, AResponseAfterTheWaiterParksWakesTheE
395436
EXPECT_EQ(opentelemetry::sdk::common::ExportResult::kSuccess, parked->finished.get());
396437
}
397438

398-
TEST_F(ElasticsearchLogsExporterSyncTests, ResponseRecordedBeforeTheWaitIsStillSeen)
439+
TEST_F(ElasticsearchLogsExporterSyncLoggingTests, ResponseRecordedBeforeTheWaitIsStillSeen)
399440
{
400441
const auto result = ExportWith([](http_client::EventHandler &handler) {
401442
FakeResponse response(200, kAcceptedBody);
@@ -477,7 +518,7 @@ std::size_t CountExporterErrors(const std::vector<std::string> &lines)
477518
// The winning side of the same rule. Without this, a recordCompletion that still records the
478519
// outcome but always answers "you did not decide it" passes every other case here while the
479520
// exporter goes silent about every failure it reports to its caller.
480-
TEST_F(ElasticsearchLogsExporterSyncTests, AWinningTerminalErrorIsReportedExactlyOnce)
521+
TEST_F(ElasticsearchLogsExporterSyncLoggingTests, AWinningTerminalErrorIsReportedExactlyOnce)
481522
{
482523
for (const auto &terminal : kErrorTerminalStates)
483524
{
@@ -511,7 +552,7 @@ TEST_F(ElasticsearchLogsExporterSyncTests, AWinningTerminalErrorIsReportedExactl
511552

512553
// The shared curl client has reported more than one terminal event for one request, see #4360,
513554
// so which of two failures is described matters rather than only how many arrive.
514-
TEST_F(ElasticsearchLogsExporterSyncTests, OnlyTheFirstTerminalFailureIsReported)
555+
TEST_F(ElasticsearchLogsExporterSyncLoggingTests, OnlyTheFirstTerminalFailureIsReported)
515556
{
516557
auto capturing = nostd::shared_ptr<internal_log::LogHandler>(new ErrorCapturingLogHandler());
517558
const auto previous = internal_log::GlobalLogHandler::GetLogHandler();
@@ -545,7 +586,7 @@ TEST_F(ElasticsearchLogsExporterSyncTests, OnlyTheFirstTerminalFailureIsReported
545586
// covers a completion recorded before the waiter arrives but never two callbacks at once. This
546587
// releases a response and a read error together. Either may win, and the point is that the result
547588
// and the diagnostic never disagree about which one did.
548-
TEST_F(ElasticsearchLogsExporterSyncTests, AResponseRacingAReadErrorAgree)
589+
TEST_F(ElasticsearchLogsExporterSyncLoggingTests, AResponseRacingAReadErrorAgree)
549590
{
550591
for (int attempt = 0; attempt < 50; ++attempt)
551592
{
@@ -607,7 +648,7 @@ TEST_F(ElasticsearchLogsExporterSyncTests, AResponseRacingAReadErrorAgree)
607648
}
608649

609650
// The losing side, over the same table rather than three of the nine.
610-
TEST_F(ElasticsearchLogsExporterSyncTests, NoTerminalErrorAfterAResponseIsReported)
651+
TEST_F(ElasticsearchLogsExporterSyncLoggingTests, NoTerminalErrorAfterAResponseIsReported)
611652
{
612653
for (const auto &terminal : kErrorTerminalStates)
613654
{
@@ -632,7 +673,7 @@ TEST_F(ElasticsearchLogsExporterSyncTests, NoTerminalErrorAfterAResponseIsReport
632673
}
633674
}
634675

635-
TEST_F(ElasticsearchLogsExporterSyncTests, ALateTerminalFailureDoesNotClaimTheExportFailed)
676+
TEST_F(ElasticsearchLogsExporterSyncLoggingTests, ALateTerminalFailureDoesNotClaimTheExportFailed)
636677
{
637678
auto capturing = nostd::shared_ptr<internal_log::LogHandler>(new ErrorCapturingLogHandler());
638679
const auto previous = internal_log::GlobalLogHandler::GetLogHandler();

0 commit comments

Comments
 (0)