Skip to content
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 3 additions & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -114,6 +114,9 @@ Increment the:
deprecated C headers (`stdint.h`, `stddef.h`, `stdlib.h`, `string.h`,
`stdio.h`, `ctype.h`, `limits.h`, `assert.h`) with their C++ equivalents
([#4349](https://github.com/open-telemetry/opentelemetry-cpp/pull/4349))
* [BUG] Stop the Elasticsearch async ForceFlush reporting success without
waiting for the sessions it was asked about
[#4337](https://github.com/open-telemetry/opentelemetry-cpp/pull/4337)

* [CONFIGURATION] Add SDK component builder interfaces to the registry
[#4358](https://github.com/open-telemetry/opentelemetry-cpp/issues/4358)
Expand Down
8 changes: 8 additions & 0 deletions exporters/elasticsearch/CMakeLists.txt
Original file line number Diff line number Diff line change
Expand Up @@ -55,4 +55,12 @@ if(OTELCPP_BUILD_TESTING)
TARGET es_log_record_exporter_test
TEST_PREFIX exporter.
TEST_LIST es_log_record_exporter_test)

# AnIndefiniteFlushParksUntilTheOutcomeArrives really does park on the
# condition variable, so a regression there does not fail, it stalls until the
# exporter's own bound expires. CTest's default is 25 minutes, which is a long
# time to spend learning that. This is deliberately well above the bounds the
# cases use themselves, so that one of them expiring reports which case failed
# rather than being killed from the outside.
set_tests_properties(${es_log_record_exporter_test} PROPERTIES TIMEOUT 60)
endif() # OTELCPP_BUILD_TESTING
Original file line number Diff line number Diff line change
Expand Up @@ -16,8 +16,9 @@

#ifdef ENABLE_ASYNC_EXPORT
# include <condition_variable>
# include <cstddef>
# include <cstdint>
# include <mutex>
# include <set>
#endif

OPENTELEMETRY_BEGIN_NAMESPACE
Expand Down Expand Up @@ -122,8 +123,16 @@ class ElasticsearchLogRecordExporter final : public opentelemetry::sdk::logs::Lo

/**
* Force flush the exporter.
*
* Waits for the asynchronous sessions already started when the call was made, bounded by the
* caller's timeout rather than by the exporter's response timeout.
*
* @param timeout an option timeout, default to max.
* @return return true when all data are exported, and false when timeout
* @return true when each of those exports has reported a terminal outcome, false on timeout.
* The outcome is published before the session is torn down, so a true return does not
* mean FinishSession() has run. Nor does it mean the batch reached Elasticsearch: a
* failed export reports too, through the internal log. Surfacing that here is
* [#3075](https://github.com/open-telemetry/opentelemetry-cpp/issues/3075).
*/
bool ForceFlush(
std::chrono::microseconds timeout = (std::chrono::microseconds::max)()) noexcept override;
Expand All @@ -149,13 +158,21 @@ class ElasticsearchLogRecordExporter final : public opentelemetry::sdk::logs::Lo
#ifdef ENABLE_ASYNC_EXPORT
struct SynchronizationData
{
std::atomic<std::size_t> session_counter_{0};
std::atomic<std::size_t> finished_session_counter_{0};
// next_session_id and running_sessions are guarded by force_flush_cv_m, so a start or a
// finish cannot interleave with a waiter's snapshot. The wait compares ids by order, so they
// must keep increasing: uint64_t rather than size_t.
std::uint64_t next_session_id{0};
std::set<std::uint64_t> running_sessions;
std::condition_variable force_flush_cv;
std::mutex force_flush_cv_m;
std::recursive_mutex force_flush_m;

// Test synchronization. Raised under force_flush_cv_m once per ForceFlush(), immediately
// after that call takes its watermark. No exporter behaviour reads it.
std::uint64_t watermarks_taken{0};
};
nostd::shared_ptr<SynchronizationData> synchronization_data_;

friend class ElasticsearchExporterTestPeer;
#endif
};
} // namespace logs
Expand Down
Loading
Loading