Expose an explicit Tracer.Flush() method

Extend the OpenTracing API with an explicit jaegertracing::Tracer::flush()
method to force spans to be flushed eagerly without closing the tracer. It
returns only when the spans are flushed.

Fixes jaegertracing#53

Call flush() from Close(), but not from the Tracer dtor. So we follow the spec
and ensure we flush buffers on explicit Close only.

Fixes jaegertracing#52

WIP. For as-yet-undiagnosed reasons `flush()` sometimes waits a full
bufferFlushInterval before returning.

Signed-off-by: Craig Ringer <[email protected]>

src/jaegertracing/Tracer.h

@@ -102,7 +102,7 @@ class Tracer : public opentracing::Tracer,
             serviceName, sampler, reporter, logger, metrics, options));
     }
 
-    ~Tracer() { Close(); }
+    ~Tracer() { close(); }
 
     std::unique_ptr<opentracing::Span>
     StartSpanWithOptions(string_view operationName,
@@ -181,6 +181,12 @@ class Tracer : public opentracing::Tracer,
     }
 
     void Close() noexcept override
+    {
+        flush();
+        close();
+    }
+
+    void close() noexcept
     {
         try {
             _reporter->close();
@@ -192,7 +198,10 @@ class Tracer : public opentracing::Tracer,
         }
     }
 
-    void close() noexcept { Close(); }
+    void flush()
+    {
+        _reporter->flush();
+    }
 
     const std::string& serviceName() const { return _serviceName; }
 

src/jaegertracing/reporters/RemoteReporter.cpp

@@ -94,7 +94,7 @@ void RemoteReporter::sweepQueue()
                 sendSpan(span);
             }
             else if (bufferFlushIntervalExpired()) {
-                flush();
+                async_flush();
             }
         } catch (...) {
             auto logger = logging::consoleLogger();
@@ -122,12 +122,13 @@ void RemoteReporter::sendSpan(const Span& span)
     }
 }
 
-void RemoteReporter::flush()
+void RemoteReporter::async_flush()
 {
     try {
         const auto flushed = _sender->flush();
         if (flushed > 0) {
             _metrics.reporterSuccess().inc(flushed);
+            _cv.notify_one();
         }
     } catch (const Transport::Exception& ex) {
         _metrics.reporterFailure().inc(ex.numFailed());
@@ -137,5 +138,14 @@ void RemoteReporter::flush()
     _lastFlush = Clock::now();
 }
 
+void RemoteReporter::flush()
+{
+    std::unique_lock<std::mutex> lock(_mutex);
+    _cv.wait(lock, [this]() {
+        async_flush();
+        return !_running || _queue.empty();
+    });
+}
+
 }  // namespace reporters
 }  // namespace jaegertracing

src/jaegertracing/reporters/RemoteReporter.h

@@ -48,12 +48,14 @@ class RemoteReporter : public Reporter {
 
     void close() override;
 
+    void flush() override;
+
   private:
     void sweepQueue();
 
     void sendSpan(const Span& span);
 
-    void flush();
+    void async_flush();
 
     bool bufferFlushIntervalExpired() const
     {

src/jaegertracing/reporters/Reporter.h

@@ -30,6 +30,8 @@ class Reporter {
     virtual void report(const Span& span) = 0;
 
     virtual void close() = 0;
+
+    virtual void flush() {};
 };
 
 }  // namespace reporters