[BUGFIX] Bound the flush, and say nothing on the way to an outcome

Eight things from review, each measured rather than accepted on description.

The async handler logged five progress states that the synchronous one has
always logged and the async one never did. The session is registered before the
request is handed to the client, and none of those events retires it, so a log
handler calling ForceFlush() from one waited for the export whose call stack it
was standing in. Every other report in this file already retires before it logs,
which is what made the omission visible: the progress cases were the one place
that broke the file's own rule. They report nothing now, and the exhaustive
switch stays so a new state is still a compile error.

A flush with no deadline of its own waited without one. The bound this path had
before the accounting was fixed is response_timeout_, which is also the value
the request itself is given, so a session outstanding when the watermark is
taken has at most that long before the client owes it a terminal event. The
bound is unchanged; what changes is that running out of it no longer reports
success. That also keeps this independent of the curl client work, where a
transfer can be left with no callback at all.

The deadline was built after the mutex was taken, so time spent waiting for the
lock came out of nobody's budget and the caller was handed a fresh one. It is
taken at entry now, and ForceFlush has a single wait path rather than two.

The GiveUpGuard could not reach its active destructor: the shutdown return
disarmed it and retired by hand, the successful path disarmed it after
SendRequest, and there was no other return between them. It is gone, and the
handler takes the completion by move rather than by copy. The invariant it was
guarding is stated where the two exits are.

WaitForWatermarks had the same thirty second bound as the whole CTest case, so a
watermark that never arrived killed the process before the case reached its own
assertion. Five seconds for the helper, sixty for CTest. The CMake comment named
a case that never parks; it names the one that does.

Concurrency between ForceFlush and Shutdown is a MUST in the stable Logs SDK
specification and this exporter had no case for it. Two now: one where the
shutdown settles the outstanding session and the flush is woken by it rather
than by its own bound, and one where the shutdown settles nothing and the flush
ends anyway.

watermarks_taken is labelled as test synchronization rather than left looking
like exporter state.

Verified by putting each defect back: a progress line restored fails
EverySessionStateIsClassifiedAndReportsAtMostOnce, the bound removed fails
ANoDeadlineFlushGivesUpAtTheResponseTimeout, and the unbounded wait restored
hangs AParkedFlushEndsEvenIfTheShutdownSettlesNothing at exit 124. All 31 cases
pass in the async build, under AddressSanitizer and under ThreadSanitizer, with
no sanitizer report and the same count in each.

Signed-off-by: thc1006 <84045975+thc1006@users.noreply.github.com>

Author: thc1006

exporters/elasticsearch/CMakeLists.txt

@@ -56,9 +56,11 @@ if(BUILD_TESTING)
     TEST_PREFIX exporter.
     TEST_LIST es_log_record_exporter_test)
 
-  # AnIndefiniteFlushReturnsOnceEverythingIsFinished exercises the branch that
-  # waits without a deadline, so a regression there does not fail, it stops.
-  # CTest's default bound is 25 minutes, which is a long time to spend learning
-  # that.
-  set_tests_properties(${es_log_record_exporter_test} PROPERTIES TIMEOUT 30)
+  # 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() # BUILD_TESTING

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

@@ -170,11 +170,12 @@ class ElasticsearchLogRecordExporter final : public opentelemetry::sdk::logs::Lo
     std::condition_variable force_flush_cv;
     std::mutex force_flush_cv_m;
 
-    // Raised once per ForceFlush(), under that mutex and immediately after the call has taken
-    // its watermark. Nothing in the exporter reads it. A case that has to start an export after
-    // a flush has snapshotted, and before it gives up, waits on this: the alternative is a sleep,
-    // which leaves the order to the scheduler, and the order that does not reproduce the defect
-    // is the one that would report a pass.
+    // Test synchronization only, and not exporter state: nothing here reads it and no behaviour
+    // depends on it. Raised once per ForceFlush(), under that mutex and immediately after the
+    // call has taken its watermark. A case that has to start an export after a flush has
+    // snapshotted, and before it gives up, waits on this. The alternative is a sleep, which
+    // leaves the order to the scheduler, and the order that does not reproduce the defect is the
+    // one that would report a pass.
     std::uint64_t watermarks_taken{0};
   };
   nostd::shared_ptr<SynchronizationData> synchronization_data_;

exporters/elasticsearch/src/es_log_record_exporter.cc

@@ -338,34 +338,31 @@ class AsyncResponseHandler : public http_client::EventHandler
     const char *failure = nullptr;
     switch (state)
     {
+      // The states a session passes through on its way to an outcome. Nothing is reported for
+      // them, and in particular nothing is logged: the log handler is replaceable, this session is
+      // registered before the request is handed to the client, and none of these events has
+      // retired it. A handler that flushed from here would wait for the session whose call stack
+      // it is standing in. Every other report in this file retires first for that reason, and
+      // there is nothing to retire on the way to an outcome.
+      case http_client::SessionState::Created:
+      case http_client::SessionState::Connecting:
+      case http_client::SessionState::Connected:
+      case http_client::SessionState::Sending:
+      // The body arrives through OnResponse(), which is what reports the outcome.
+      case http_client::SessionState::Response:
+        break;
       case http_client::SessionState::CreateFailed:
         failure = "[ES Log Exporter] Create request to elasticsearch failed";
         break;
-      case http_client::SessionState::Created:
-        OTEL_INTERNAL_LOG_DEBUG("[ES Log Exporter] Session created");
-        break;
       case http_client::SessionState::Destroyed:
         failure = "[ES Log Exporter] Session to elasticsearch destroyed before a response";
         break;
-      case http_client::SessionState::Connecting:
-        OTEL_INTERNAL_LOG_DEBUG("[ES Log Exporter] Connecting to elasticsearch");
-        break;
       case http_client::SessionState::ConnectFailed:
         failure = "[ES Log Exporter] Connection to elasticsearch failed";
         break;
-      case http_client::SessionState::Connected:
-        OTEL_INTERNAL_LOG_DEBUG("[ES Log Exporter] Connected to elasticsearch");
-        break;
-      case http_client::SessionState::Sending:
-        OTEL_INTERNAL_LOG_DEBUG("[ES Log Exporter] Sending request to elasticsearch");
-        break;
       case http_client::SessionState::SendFailed:
         failure = "[ES Log Exporter] Request failed to be sent to elasticsearch";
         break;
-      case http_client::SessionState::Response:
-        // The body arrives through OnResponse(), which is what reports the outcome.
-        OTEL_INTERNAL_LOG_DEBUG("[ES Log Exporter] Response received from elasticsearch");
-        break;
       case http_client::SessionState::SSLHandshakeFailed:
         failure = "[ES Log Exporter] SSL handshake to elasticsearch failed";
         break;
@@ -482,10 +479,9 @@ sdk::common::ExportResult ElasticsearchLogRecordExporter::Export(
     //
     // That is the whole of what the ordering buys, and it is worth being exact about the rest.
     // It does not make a log handler safe to re-enter the exporter from in general: one that
-    // calls ForceFlush() without a deadline from a progress event blocks the Export() that has
-    // not handed its request to the client yet, and one that flushes from any callback the HTTP
-    // client dispatches can wait on work only that client thread can advance. Neither is new
-    // here and neither is fixed here.
+    // flushes from any callback the HTTP client dispatches can still wait on work that only the
+    // client thread it is standing on can advance. That is
+    // https://github.com/open-telemetry/opentelemetry-cpp/issues/4435, and it is not fixed here.
     if (result != opentelemetry::sdk::common::ExportResult::kSuccess)
     {
       OTEL_INTERNAL_LOG_ERROR("[ES Log Exporter] ERROR: Export "
@@ -500,37 +496,19 @@ sdk::common::ExportResult ElasticsearchLogRecordExporter::Export(
     return true;
   };
 
-  // A return added between here and SendRequest() would otherwise strand a waiter on a session
-  // that can never finish. Reporting through the same completion leaves a session one way out,
-  // and says so in the log rather than dropping the batch silently.
-  struct GiveUpGuard
-  {
-    Completion *report = nullptr;
-
-    GiveUpGuard()                               = default;
-    GiveUpGuard(const GiveUpGuard &)            = delete;
-    GiveUpGuard &operator=(const GiveUpGuard &) = delete;
-    GiveUpGuard(GiveUpGuard &&)                 = delete;
-    GiveUpGuard &operator=(GiveUpGuard &&)      = delete;
-
-    ~GiveUpGuard()
-    {
-      if (report != nullptr)
-      {
-        (*report)(opentelemetry::sdk::common::ExportResult::kFailure);
-      }
-    }
-  } guard;
-  guard.report = &complete;
+  // Two ways out from here, and each one is spelled out below: the shutdown refusal retires by
+  // hand, and the handler takes the session over once SendRequest() has it. A return added
+  // between them has to do one or the other, or it strands a waiter on a session that can never
+  // finish.
 #endif
 
   // Return failure if this exporter has been shutdown
   if (isShutdown())
   {
 #ifdef ENABLE_ASYNC_EXPORT
-    // Retired before anything replaceable runs, and reported here rather than through the guard,
-    // so a log handler that flushes does not wait for this Export() and one refusal reads as one.
-    guard.report = nullptr;
+    // Retired before anything replaceable runs, and reported by the line below rather than
+    // through the completion, so a log handler that flushes does not wait for this Export() and
+    // one refusal reads as one.
     retire();
 #endif
     OTEL_INTERNAL_LOG_ERROR("[ES Log Exporter] Exporting "
@@ -573,10 +551,11 @@ sdk::common::ExportResult ElasticsearchLogRecordExporter::Export(
 
 #ifdef ENABLE_ASYNC_EXPORT
   // Send the request
-  auto handler = std::make_shared<AsyncResponseHandler>(session, Completion(complete),
-                                                        options_.console_debug_);
+  // The handler reports this session from here on, so the completion goes to it rather than
+  // being copied to it.
+  auto handler =
+      std::make_shared<AsyncResponseHandler>(session, std::move(complete), options_.console_debug_);
   session->SendRequest(handler);
-  guard.report = nullptr;  // the handler reports this session from here on
   return sdk::common::ExportResult::kSuccess;
 #else
   // Send the request
@@ -627,6 +606,25 @@ bool ElasticsearchLogRecordExporter::ForceFlush(std::chrono::microseconds timeou
   timeout = opentelemetry::common::DurationUtil::AdjustWaitForTimeout(
       timeout, std::chrono::microseconds::zero());
 
+  // A caller that asks for no deadline still gets one, because a flush that cannot end is worse
+  // than one that reports it did not finish, and the client is not obliged to call back at all.
+  // response_timeout_ is the bound this wait already had before the accounting was fixed, and it
+  // is the same value the request itself is given, so a session outstanding when the watermark is
+  // taken has at most that long before the client owes it a terminal event. The bound is
+  // unchanged; what changes is that running out of it no longer reports success.
+  if (timeout <= std::chrono::microseconds::zero())
+  {
+    timeout = std::chrono::seconds{options_.response_timeout_};
+  }
+
+  // Taken before the lock rather than after it, so the time spent waiting for the mutex comes out
+  // of the caller's budget instead of being handed back as a fresh one. A plain mutex still
+  // cannot promise a bound, but acquiring it should not add a second full timeout to the one that
+  // was asked for.
+  const std::chrono::steady_clock::time_point deadline =
+      std::chrono::steady_clock::now() +
+      std::chrono::duration_cast<std::chrono::steady_clock::duration>(timeout);
+
   std::unique_lock<std::mutex> lock(synchronization_data_->force_flush_cv_m);
 
   // The snapshot is the next id, not a count: a session started after it takes a larger id and
@@ -639,20 +637,9 @@ bool ElasticsearchLogRecordExporter::ForceFlush(std::chrono::microseconds timeou
     return running.empty() || *running.begin() >= watermark;
   };
 
-  if (timeout <= std::chrono::microseconds::zero())
-  {
-    // wait() only returns once the predicate holds, so the flush has completed.
-    synchronization_data_->force_flush_cv.wait(lock, flushed);
-    return true;
-  }
-
   // One deadline for the call, so a wakeup that is not a completion resumes against what is left
   // rather than starting the wait again. wait_until() returns the predicate, so a flush that ran
   // out of time cannot report success.
-  const std::chrono::steady_clock::time_point deadline =
-      std::chrono::steady_clock::now() +
-      std::chrono::duration_cast<std::chrono::steady_clock::duration>(timeout);
-
   return synchronization_data_->force_flush_cv.wait_until(lock, deadline, flushed);
 #else
   return true;