Skip to content

Commit 0109b5c

Browse files
committed
[BUG] Bound the Elasticsearch async ForceFlush and decide it by session identity
ForceFlush waited for options_.response_timeout_ rather than the caller's timeout, and the timeout branch returned true whatever had happened, so every flush that ran out of time reported success. It also compared monotonic totals with no session identity, so a completion from a session started after the call could satisfy a waiter for one started before it, and a batch already inside Export() was not waited for at all. One steady_clock deadline taken at entry, and a wait on a predicate over the set of running session ids against a watermark. wait_until returns the predicate, so the answer is the caller's question rather than a leftover duration. The serialising lock is gone: each call snapshots what it waits for and publishes nothing, so two callers were already safe side by side. Stacked on #4502, which carries the exactly-once completion this accounting depends on. That half was extracted so each pull request closes one issue and is reviewable on its own. Signed-off-by: thc1006 <84045975+thc1006@users.noreply.github.com>
1 parent 9d5224c commit 0109b5c

5 files changed

Lines changed: 801 additions & 60 deletions

File tree

‎CHANGELOG.md‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -131,6 +131,9 @@ Increment the:
131131
deprecated C headers (`stdint.h`, `stddef.h`, `stdlib.h`, `string.h`,
132132
`stdio.h`, `ctype.h`, `limits.h`, `assert.h`) with their C++ equivalents
133133
([#4349](https://github.com/open-telemetry/opentelemetry-cpp/pull/4349))
134+
* [BUG] Stop the Elasticsearch async ForceFlush reporting success without
135+
waiting for the sessions it was asked about
136+
[#4337](https://github.com/open-telemetry/opentelemetry-cpp/pull/4337)
134137

135138
* [CONFIGURATION] Add SDK component builder interfaces to the registry
136139
[#4358](https://github.com/open-telemetry/opentelemetry-cpp/issues/4358)

‎exporters/elasticsearch/CMakeLists.txt‎

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -55,4 +55,12 @@ if(OTELCPP_BUILD_TESTING)
5555
TARGET es_log_record_exporter_test
5656
TEST_PREFIX exporter.
5757
TEST_LIST es_log_record_exporter_test)
58+
59+
# AnIndefiniteFlushParksUntilTheOutcomeArrives really does park on the
60+
# condition variable, so a regression there does not fail, it stalls until the
61+
# exporter's own bound expires. CTest's default is 25 minutes, which is a long
62+
# time to spend learning that. This is deliberately well above the bounds the
63+
# cases use themselves, so that one of them expiring reports which case failed
64+
# rather than being killed from the outside.
65+
set_tests_properties(${es_log_record_exporter_test} PROPERTIES TIMEOUT 60)
5866
endif() # OTELCPP_BUILD_TESTING

‎exporters/elasticsearch/include/opentelemetry/exporters/elasticsearch/es_log_record_exporter.h‎

Lines changed: 22 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -16,8 +16,9 @@
1616

1717
#ifdef ENABLE_ASYNC_EXPORT
1818
# include <condition_variable>
19-
# include <cstddef>
19+
# include <cstdint>
2020
# include <mutex>
21+
# include <set>
2122
#endif
2223

2324
OPENTELEMETRY_BEGIN_NAMESPACE
@@ -122,8 +123,16 @@ class ElasticsearchLogRecordExporter final : public opentelemetry::sdk::logs::Lo
122123

123124
/**
124125
* Force flush the exporter.
126+
*
127+
* Waits for the asynchronous sessions already started when the call was made, bounded by the
128+
* caller's timeout rather than by the exporter's response timeout.
129+
*
125130
* @param timeout an option timeout, default to max.
126-
* @return return true when all data are exported, and false when timeout
131+
* @return true when each of those exports has reported a terminal outcome, false on timeout.
132+
* The outcome is published before the session is torn down, so a true return does not
133+
* mean FinishSession() has run. Nor does it mean the batch reached Elasticsearch: a
134+
* failed export reports too, through the internal log. Surfacing that here is
135+
* [#3075](https://github.com/open-telemetry/opentelemetry-cpp/issues/3075).
127136
*/
128137
bool ForceFlush(
129138
std::chrono::microseconds timeout = (std::chrono::microseconds::max)()) noexcept override;
@@ -149,13 +158,21 @@ class ElasticsearchLogRecordExporter final : public opentelemetry::sdk::logs::Lo
149158
#ifdef ENABLE_ASYNC_EXPORT
150159
struct SynchronizationData
151160
{
152-
std::atomic<std::size_t> session_counter_{0};
153-
std::atomic<std::size_t> finished_session_counter_{0};
161+
// next_session_id and running_sessions are guarded by force_flush_cv_m, so a start or a
162+
// finish cannot interleave with a waiter's snapshot. The wait compares ids by order, so they
163+
// must keep increasing: uint64_t rather than size_t.
164+
std::uint64_t next_session_id{0};
165+
std::set<std::uint64_t> running_sessions;
154166
std::condition_variable force_flush_cv;
155167
std::mutex force_flush_cv_m;
156-
std::recursive_mutex force_flush_m;
168+
169+
// Test synchronization. Raised under force_flush_cv_m once per ForceFlush(), immediately
170+
// after that call takes its watermark. No exporter behaviour reads it.
171+
std::uint64_t watermarks_taken{0};
157172
};
158173
nostd::shared_ptr<SynchronizationData> synchronization_data_;
174+
175+
friend class ElasticsearchExporterTestPeer;
159176
#endif
160177
};
161178
} // namespace logs

‎exporters/elasticsearch/src/es_log_record_exporter.cc‎

Lines changed: 95 additions & 55 deletions
Original file line numberDiff line numberDiff line change
@@ -5,7 +5,7 @@
55
#include <chrono>
66
#include <condition_variable>
77
#include <cstdint>
8-
#include <map>
8+
#include <map> // IWYU pragma: keep
99
#include <memory> // IWYU pragma: keep
1010
#include <mutex>
1111
#include <nlohmann/json.hpp>
@@ -27,9 +27,13 @@
2727
#include "opentelemetry/sdk/logs/recordable.h"
2828
#include "opentelemetry/version.h"
2929

30+
// Half this file only exists under ENABLE_ASYNC_EXPORT, and include-what-you-use asks for these
31+
// in the configurations that build it and asks for them to go in the ones that do not.
3032
#ifdef ENABLE_ASYNC_EXPORT
3133
# include <cstddef>
3234
# include <functional>
35+
# include <set>
36+
3337
# include "opentelemetry/common/timestamp.h"
3438
#endif
3539

@@ -423,12 +427,7 @@ ElasticsearchLogRecordExporter::ElasticsearchLogRecordExporter(
423427
,
424428
synchronization_data_(new SynchronizationData())
425429
#endif
426-
{
427-
#ifdef ENABLE_ASYNC_EXPORT
428-
synchronization_data_->finished_session_counter_.store(0);
429-
synchronization_data_->session_counter_.store(0);
430-
#endif
431-
}
430+
{}
432431

433432
std::unique_ptr<sdklogs::Recordable> ElasticsearchLogRecordExporter::MakeRecordable() noexcept
434433
{
@@ -438,9 +437,72 @@ std::unique_ptr<sdklogs::Recordable> ElasticsearchLogRecordExporter::MakeRecorda
438437
sdk::common::ExportResult ElasticsearchLogRecordExporter::Export(
439438
const nostd::span<std::unique_ptr<sdklogs::Recordable>> &records) noexcept
440439
{
440+
#ifdef ENABLE_ASYNC_EXPORT
441+
// Registered before anything that can return, so a flush asked from the moment these records
442+
// arrive waits for them, and every exit below reports through the guard.
443+
const std::size_t span_count = records.size();
444+
auto synchronization_data = synchronization_data_;
445+
446+
std::uint64_t session_id = 0;
447+
{
448+
std::lock_guard<std::mutex> lock(synchronization_data_->force_flush_cv_m);
449+
session_id = synchronization_data_->next_session_id++;
450+
synchronization_data_->running_sessions.insert(session_id);
451+
}
452+
453+
// Retiring the export is separate from describing what happened to it, because the refusal
454+
// below has to retire without having an outcome to report through the completion.
455+
auto retire = [session_id, synchronization_data]() noexcept {
456+
bool retired = false;
457+
{
458+
// Published under the mutex ForceFlush() waits on. A waiter that has evaluated its
459+
// predicate but not yet parked would otherwise not see this until the next wakeup.
460+
std::lock_guard<std::mutex> lock(synchronization_data->force_flush_cv_m);
461+
retired = synchronization_data->running_sessions.erase(session_id) == 1;
462+
}
463+
synchronization_data->force_flush_cv.notify_all();
464+
return retired;
465+
};
466+
467+
using Completion = std::function<bool(opentelemetry::sdk::common::ExportResult)>;
468+
Completion complete = [span_count,
469+
retire](opentelemetry::sdk::common::ExportResult result) noexcept {
470+
retire();
471+
472+
// Logged after the session is retired. The log handler is replaceable, and one that calls
473+
// ForceFlush() would otherwise wait for the very session this call has not let go of yet.
474+
//
475+
// That is the whole of what the ordering buys, and it is worth being exact about the rest.
476+
// It does not make a log handler safe to re-enter the exporter from in general: one that
477+
// flushes from any callback the HTTP client dispatches can still wait on work that only the
478+
// client thread it is standing on can advance. That is
479+
// https://github.com/open-telemetry/opentelemetry-cpp/issues/4435, and it is not fixed here.
480+
if (result != opentelemetry::sdk::common::ExportResult::kSuccess)
481+
{
482+
OTEL_INTERNAL_LOG_ERROR("[ES Log Exporter] ERROR: Export "
483+
<< span_count
484+
<< " log record(s) error: " << static_cast<int>(result));
485+
}
486+
else
487+
{
488+
OTEL_INTERNAL_LOG_DEBUG("[ES Log Exporter] Export " << span_count
489+
<< " log record(s) success");
490+
}
491+
return true;
492+
};
493+
494+
// Exactly two exits below: the shutdown refusal retires by hand, and the handler takes the
495+
// session over after SendRequest(). A return added between them must do one or the other.
496+
#endif
497+
441498
// Return failure if this exporter has been shutdown
442499
if (isShutdown())
443500
{
501+
#ifdef ENABLE_ASYNC_EXPORT
502+
// Retired before the log line below, so a handler that flushes does not wait on this
503+
// Export(). The refusal is reported there, not through the completion.
504+
retire();
505+
#endif
444506
OTEL_INTERNAL_LOG_ERROR("[ES Log Exporter] Exporting "
445507
<< records.size() << " log(s) failed, exporter is shutdown");
446508
return sdk::common::ExportResult::kFailure;
@@ -487,29 +549,10 @@ sdk::common::ExportResult ElasticsearchLogRecordExporter::Export(
487549

488550
#ifdef ENABLE_ASYNC_EXPORT
489551
// Send the request
490-
synchronization_data_->session_counter_.fetch_add(1, std::memory_order_release);
491-
std::size_t span_count = records.size();
492-
auto synchronization_data = synchronization_data_;
493-
auto handler = std::make_shared<AsyncResponseHandler>(
494-
session,
495-
[span_count, synchronization_data](opentelemetry::sdk::common::ExportResult result) {
496-
if (result != opentelemetry::sdk::common::ExportResult::kSuccess)
497-
{
498-
OTEL_INTERNAL_LOG_ERROR("[ES Log Exporter] ERROR: Export "
499-
<< span_count
500-
<< " log record(s) error: " << static_cast<int>(result));
501-
}
502-
else
503-
{
504-
OTEL_INTERNAL_LOG_DEBUG("[ES Log Exporter] Export " << span_count
505-
<< " log record(s) success");
506-
}
507-
508-
synchronization_data->finished_session_counter_.fetch_add(1, std::memory_order_release);
509-
synchronization_data->force_flush_cv.notify_all();
510-
return true;
511-
},
512-
options_.console_debug_);
552+
// The handler reports this session from here on, so the completion goes to it rather than
553+
// being copied to it.
554+
auto handler =
555+
std::make_shared<AsyncResponseHandler>(session, std::move(complete), options_.console_debug_);
513556
session->SendRequest(handler);
514557
return sdk::common::ExportResult::kSuccess;
515558
#else
@@ -554,41 +597,38 @@ bool ElasticsearchLogRecordExporter::ForceFlush(std::chrono::microseconds timeou
554597
OPENTELEMETRY_MAYBE_UNUSED) noexcept
555598
{
556599
#ifdef ENABLE_ASYNC_EXPORT
557-
std::lock_guard<std::recursive_mutex> lock_guard{synchronization_data_->force_flush_m};
558-
std::size_t running_counter =
559-
synchronization_data_->session_counter_.load(std::memory_order_acquire);
560600
// ASAN will report chrono: runtime error: signed integer overflow: A + B cannot be represented
561601
// in type 'long int' here. So we reset timeout to meet signed long int limit here.
602+
// Returns zero for a timeout not worth waiting against, which is also how a caller asks for
603+
// no deadline.
562604
timeout = opentelemetry::common::DurationUtil::AdjustWaitForTimeout(
563605
timeout, std::chrono::microseconds::zero());
564606

565-
std::chrono::steady_clock::duration timeout_steady =
566-
std::chrono::duration_cast<std::chrono::steady_clock::duration>(timeout);
567-
if (timeout_steady <= std::chrono::steady_clock::duration::zero())
607+
// No deadline still gets one, since a client is not obliged to call back. response_timeout_
608+
// is the value the request itself carries, so an outstanding session owes an event within it.
609+
if (timeout <= std::chrono::microseconds::zero())
568610
{
569-
timeout_steady = (std::chrono::steady_clock::duration::max)();
611+
timeout = std::chrono::seconds{options_.response_timeout_};
570612
}
571613

572-
std::unique_lock<std::mutex> lk_cv(synchronization_data_->force_flush_cv_m);
573-
// Wait for all the sessions to finish
574-
while (timeout_steady > std::chrono::steady_clock::duration::zero())
575-
{
576-
if (synchronization_data_->finished_session_counter_.load(std::memory_order_acquire) >=
577-
running_counter)
578-
{
579-
break;
580-
}
614+
// Before the lock, so acquiring the mutex spends the caller's budget and not a fresh one.
615+
const std::chrono::steady_clock::time_point deadline =
616+
std::chrono::steady_clock::now() +
617+
std::chrono::duration_cast<std::chrono::steady_clock::duration>(timeout);
581618

582-
std::chrono::steady_clock::time_point start_timepoint = std::chrono::steady_clock::now();
583-
if (std::cv_status::no_timeout != synchronization_data_->force_flush_cv.wait_for(
584-
lk_cv, std::chrono::seconds{options_.response_timeout_}))
585-
{
586-
break;
587-
}
588-
timeout_steady -= std::chrono::steady_clock::now() - start_timepoint;
589-
}
619+
std::unique_lock<std::mutex> lock(synchronization_data_->force_flush_cv_m);
620+
621+
// The next id, not a count: a session started later takes a larger id and cannot stand in for
622+
// one of these. Ids are issued in order, so the smallest one still running decides.
623+
const std::uint64_t watermark = synchronization_data_->next_session_id;
624+
++synchronization_data_->watermarks_taken;
625+
const auto flushed = [this, watermark]() {
626+
const auto &running = synchronization_data_->running_sessions;
627+
return running.empty() || *running.begin() >= watermark;
628+
};
590629

591-
return timeout_steady > std::chrono::steady_clock::duration::zero();
630+
// wait_until() returns the predicate, so running out of time cannot report success.
631+
return synchronization_data_->force_flush_cv.wait_until(lock, deadline, flushed);
592632
#else
593633
return true;
594634
#endif

0 commit comments

Comments
 (0)