|
24 | 24 | #include <cstddef> |
25 | 25 | #include <cstdint> |
26 | 26 | #include <functional> |
| 27 | +#include <future> |
27 | 28 | #include <initializer_list> |
28 | 29 | #include <mutex> |
29 | 30 | #include <string> |
@@ -228,6 +229,53 @@ class FakeSession : public http_client::Session |
228 | 229 | EventScript script_; |
229 | 230 | }; |
230 | 231 |
|
| 232 | +// Keeps the handler and returns, so the export reaches its wait with nothing recorded and only a |
| 233 | +// notification can end it. |
| 234 | +class DeferredSession : public http_client::Session |
| 235 | +{ |
| 236 | +public: |
| 237 | + DeferredSession(std::shared_ptr<http_client::EventHandler> *slot, std::promise<void> *arrived) |
| 238 | + : slot_(slot), arrived_(arrived) |
| 239 | + {} |
| 240 | + |
| 241 | + std::shared_ptr<http_client::Request> CreateRequest() noexcept override |
| 242 | + { |
| 243 | + return std::make_shared<FakeRequest>(); |
| 244 | + } |
| 245 | + void SendRequest(std::shared_ptr<http_client::EventHandler> handler) noexcept override |
| 246 | + { |
| 247 | + *slot_ = std::move(handler); |
| 248 | + arrived_->set_value(); |
| 249 | + } |
| 250 | + bool IsSessionActive() noexcept override { return true; } |
| 251 | + bool CancelSession() noexcept override { return true; } |
| 252 | + bool FinishSession() noexcept override { return true; } |
| 253 | + |
| 254 | +private: |
| 255 | + std::shared_ptr<http_client::EventHandler> *slot_; |
| 256 | + std::promise<void> *arrived_; |
| 257 | +}; |
| 258 | + |
| 259 | +class DeferredHttpClient : public http_client::HttpClient |
| 260 | +{ |
| 261 | +public: |
| 262 | + DeferredHttpClient(std::shared_ptr<http_client::EventHandler> *slot, std::promise<void> *arrived) |
| 263 | + : slot_(slot), arrived_(arrived) |
| 264 | + {} |
| 265 | + |
| 266 | + std::shared_ptr<http_client::Session> CreateSession(nostd::string_view) noexcept override |
| 267 | + { |
| 268 | + return std::make_shared<DeferredSession>(slot_, arrived_); |
| 269 | + } |
| 270 | + bool CancelAllSessions() noexcept override { return true; } |
| 271 | + bool FinishAllSessions() noexcept override { return true; } |
| 272 | + void SetMaxSessionsPerConnection(std::size_t) noexcept override {} |
| 273 | + |
| 274 | +private: |
| 275 | + std::shared_ptr<http_client::EventHandler> *slot_; |
| 276 | + std::promise<void> *arrived_; |
| 277 | +}; |
| 278 | + |
231 | 279 | class FakeHttpClient : public http_client::HttpClient |
232 | 280 | { |
233 | 281 | public: |
@@ -273,6 +321,80 @@ class ElasticsearchLogsExporterSyncTests : public ::testing::Test |
273 | 321 | }; |
274 | 322 | } // namespace |
275 | 323 |
|
| 324 | +namespace |
| 325 | +{ |
| 326 | +// Runs one export on another thread against a session that keeps its handler, and hands back the |
| 327 | +// handler once the exporter has reached it. Everything is held by shared_ptr and the thread is |
| 328 | +// detached, because waitForResponse has no deadline: if a wake-up stops working the export never |
| 329 | +// returns, and joining it would hang the binary rather than fail the case. |
| 330 | +struct ParkedExport |
| 331 | +{ |
| 332 | + std::shared_ptr<DeferredHttpClient> client; |
| 333 | + std::shared_ptr<logs_exporter::ElasticsearchLogRecordExporter> exporter; |
| 334 | + std::shared_ptr<http_client::EventHandler> handler; |
| 335 | + std::promise<void> arrived; |
| 336 | + std::promise<opentelemetry::sdk::common::ExportResult> done; |
| 337 | + std::future<opentelemetry::sdk::common::ExportResult> finished; |
| 338 | +}; |
| 339 | + |
| 340 | +std::shared_ptr<ParkedExport> StartParkedExport() |
| 341 | +{ |
| 342 | + auto parked = std::make_shared<ParkedExport>(); |
| 343 | + parked->client = std::make_shared<DeferredHttpClient>(&parked->handler, &parked->arrived); |
| 344 | + |
| 345 | + logs_exporter::ElasticsearchExporterOptions options; |
| 346 | + parked->exporter = |
| 347 | + std::make_shared<logs_exporter::ElasticsearchLogRecordExporter>(options, parked->client); |
| 348 | + |
| 349 | + auto reached = parked->arrived.get_future(); |
| 350 | + auto finished = parked->done.get_future(); |
| 351 | + |
| 352 | + std::thread([parked] { |
| 353 | + auto record = parked->exporter->MakeRecordable(); |
| 354 | + parked->done.set_value( |
| 355 | + parked->exporter->Export(nostd::span<std::unique_ptr<sdklogs::Recordable>>(&record, 1))); |
| 356 | + }).detach(); |
| 357 | + |
| 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"; |
| 363 | + |
| 364 | + parked->finished = std::move(finished); |
| 365 | + return parked; |
| 366 | +} |
| 367 | +} // namespace |
| 368 | + |
| 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) |
| 372 | +{ |
| 373 | + auto parked = StartParkedExport(); |
| 374 | + ASSERT_TRUE(parked->handler); |
| 375 | + |
| 376 | + parked->handler->OnEvent(http_client::SessionState::ReadError, ""); |
| 377 | + |
| 378 | + ASSERT_EQ(std::future_status::ready, parked->finished.wait_for(std::chrono::seconds{10})) |
| 379 | + << "the read error never woke the export"; |
| 380 | + EXPECT_EQ(opentelemetry::sdk::common::ExportResult::kFailure, parked->finished.get()); |
| 381 | +} |
| 382 | + |
| 383 | +// The same for the success half, which also holds that the wait is a wait: a waitForResponse that |
| 384 | +// only read the current state would answer before this response arrives. |
| 385 | +TEST_F(ElasticsearchLogsExporterSyncTests, AResponseAfterTheWaiterParksWakesTheExport) |
| 386 | +{ |
| 387 | + auto parked = StartParkedExport(); |
| 388 | + ASSERT_TRUE(parked->handler); |
| 389 | + |
| 390 | + FakeResponse response(200, kAcceptedBody); |
| 391 | + parked->handler->OnResponse(response); |
| 392 | + |
| 393 | + ASSERT_EQ(std::future_status::ready, parked->finished.wait_for(std::chrono::seconds{10})) |
| 394 | + << "the response never woke the export"; |
| 395 | + EXPECT_EQ(opentelemetry::sdk::common::ExportResult::kSuccess, parked->finished.get()); |
| 396 | +} |
| 397 | + |
276 | 398 | TEST_F(ElasticsearchLogsExporterSyncTests, ResponseRecordedBeforeTheWaitIsStillSeen) |
277 | 399 | { |
278 | 400 | const auto result = ExportWith([](http_client::EventHandler &handler) { |
@@ -337,7 +459,7 @@ const ErrorTerminalState kErrorTerminalStates[] = { |
337 | 459 | {http_client::SessionState::Cancelled, "(manually) cancelled"}, |
338 | 460 | }; |
339 | 461 |
|
340 | | -// Counts the exporter's own error lines, and returns the ones that carry a given message. |
| 462 | +// Counts the exporter's own error lines. |
341 | 463 | std::size_t CountExporterErrors(const std::vector<std::string> &lines) |
342 | 464 | { |
343 | 465 | std::size_t count = 0; |
|
0 commit comments