ensure all span closed before sending span (#50)

* ensure all span closed before sending span

* ci

Author: Shikugawa

bazel/repositories.bzl

@@ -11,10 +11,9 @@ def cpp2sky_dependencies():
 def skywalking_data_collect_protocol():
   http_archive(
     name = "skywalking_data_collect_protocol",
-    sha256 = "ca495537bb85dbe8df5984ac7b571a0b87660281be69be1f9a0fa16fbf58f953",
-    # TODO(shikugawa): Bazel upstreaming
-    urls = ["https://github.com/Shikugawa/skywalking-data-collect-protocol/archive/v8.1.0-bazel.tar.gz"],
-    strip_prefix = "skywalking-data-collect-protocol-8.1.0-bazel",
+    sha256 = "8158e095b9b37c39e18938fe08bd4fca6f2b8e16763ff21fe8118d79241a6e0b",
+    urls = ["https://github.com/apache/skywalking-data-collect-protocol/archive/v8.3.0.tar.gz"],
+    strip_prefix = "skywalking-data-collect-protocol-8.3.0",
   )
 
 def com_github_grpc_grpc():

cpp2sky/internal/async_client.h

@@ -87,7 +87,6 @@ enum class Operation : uint8_t {
   Connected = 1,
   Idle = 2,
   WriteDone = 3,
-  Finished = 4,
 };
 
 template <class RequestType, class ResponseType>
@@ -116,7 +115,7 @@ class AsyncStream {
   /**
    * Handle incoming event related to this stream.
    */
-  virtual bool handleOperation(Operation incoming_op) = 0;
+  virtual void handleOperation(Operation incoming_op) = 0;
 };
 
 template <class RequestType>

cpp2sky/segment_context.h

@@ -262,6 +262,11 @@ class SegmentContext {
    * Whether belonging span can be skipped analysis or not.
    */
   virtual bool skipAnalysis() = 0;
+
+  /**
+   * Determine whether to send this segment or not.
+   */
+  virtual bool readyToSend() = 0;
 };
 
 using SegmentContextPtr = std::shared_ptr<SegmentContext>;

cpp2sky/tracer.h

@@ -33,6 +33,6 @@ class Tracer {
 
 using TracerPtr = std::unique_ptr<Tracer>;
 
-TracerPtr createInsecureGrpcTracer(TracerConfig& cfg);
+TracerPtr createInsecureGrpcTracer(const TracerConfig& cfg);
 
 }  // namespace cpp2sky

source/grpc_async_client_impl.cc

@@ -82,7 +82,7 @@ GrpcAsyncSegmentReporterClient::~GrpcAsyncSegmentReporterClient() {
   // failed to send message and close stream, then recreate new stream and try
   // to do it. This process will continue forever without sending explicit
   // signal.
-  {
+  if (stream_) {
     std::unique_lock<std::mutex> lck(mux_);
     while (!drained_messages_.empty()) {
       cv_.wait(lck);
@@ -149,9 +149,6 @@ GrpcAsyncSegmentReporterStream::~GrpcAsyncSegmentReporterStream() {
     }
   }
   gpr_log(GPR_INFO, "%ld pending messages drained.", pending_messages_size);
-
-  ctx_.TryCancel();
-  request_writer_->Finish(&status_, toTag(&finish_));
 }
 
 bool GrpcAsyncSegmentReporterStream::startStream() {
@@ -184,7 +181,7 @@ bool GrpcAsyncSegmentReporterStream::clearPendingMessages() {
   return true;
 }
 
-bool GrpcAsyncSegmentReporterStream::handleOperation(Operation incoming_op) {
+void GrpcAsyncSegmentReporterStream::handleOperation(Operation incoming_op) {
   state_ = incoming_op;
   if (state_ == Operation::Connected) {
     gpr_log(GPR_INFO, "Established connection: %s",
@@ -209,16 +206,9 @@ bool GrpcAsyncSegmentReporterStream::handleOperation(Operation incoming_op) {
     if (pending_messages_.empty()) {
       cv_.notify_all();
     }
-    return true;
-  } else if (state_ == Operation::Finished) {
-    gpr_log(GPR_INFO, "Stream closed with http status: %d",
-            grpcStatusToGenericHttpStatus(status_.error_code()));
-    if (!status_.ok()) {
-      gpr_log(GPR_ERROR, "%s", status_.error_message().c_str());
-    }
-    return false;
+  } else {
+    throw TracerException("Unknown stream operation");
   }
-  throw TracerException("Unknown stream operation");
 }
 
 AsyncStreamPtr<TracerRequestType> GrpcAsyncSegmentReporterStreamFactory::create(

source/grpc_async_client_impl.h

@@ -112,7 +112,7 @@ class GrpcAsyncSegmentReporterStream final
   // AsyncStream
   bool startStream() override;
   void sendMessage(TracerRequestType message) override;
-  bool handleOperation(Operation incoming_op) override;
+  void handleOperation(Operation incoming_op) override;
   void undrainMessage(TracerRequestType message) override {
     pending_messages_.push(message);
   }
@@ -122,7 +122,6 @@ class GrpcAsyncSegmentReporterStream final
 
   AsyncClient<TracerRequestType, TracerResponseType>* client_;
   TracerResponseType commands_;
-  grpc::Status status_;
   grpc::ClientContext ctx_;
   std::unique_ptr<grpc::ClientAsyncWriter<TracerRequestType>> request_writer_;
   CircularBuffer<TracerRequestType> pending_messages_{
@@ -131,7 +130,6 @@ class GrpcAsyncSegmentReporterStream final
 
   TaggedStream connected_{Operation::Connected, this};
   TaggedStream write_done_{Operation::WriteDone, this};
-  TaggedStream finish_{Operation::Finished, this};
 
   std::condition_variable& cv_;
 };

source/segment_context_impl.cc

@@ -122,15 +122,16 @@ void CurrentSegmentSpanImpl::endSpan() {
   assert(!finished_);
   auto now = TimePoint<SystemTime>();
   endSpan(now);
-  finished_ = true;
 }
 
 void CurrentSegmentSpanImpl::endSpan(TimePoint<SystemTime> current_time) {
   end_time_ = current_time.fetch();
+  finished_ = true;
 }
 
 void CurrentSegmentSpanImpl::endSpan(TimePoint<SteadyTime> current_time) {
   end_time_ = current_time.fetch();
+  finished_ = true;
 }
 
 void CurrentSegmentSpanImpl::setComponentId(int32_t component_id) {
@@ -251,6 +252,15 @@ SegmentObject SegmentContextImpl::createSegmentObject() {
   return obj;
 }
 
+bool SegmentContextImpl::readyToSend() {
+  for (const auto& span : spans_) {
+    if (!span->finished()) {
+      return false;
+    }
+  }
+  return true;
+}
+
 SegmentContextFactoryImpl::SegmentContextFactoryImpl(const TracerConfig& cfg)
     : service_name_(cfg.service_name()), instance_name_(cfg.instance_name()) {}
 

source/segment_context_impl.h

@@ -161,6 +161,7 @@ class SegmentContextImpl : public SegmentContext {
   SegmentObject createSegmentObject() override;
   void setSkipAnalysis() override { should_skip_analysis_ = true; }
   bool skipAnalysis() override { return should_skip_analysis_; }
+  bool readyToSend() override;
 
  private:
   std::string encodeSpan(CurrentSegmentSpanPtr parent_span,

source/tracer_impl.cc

@@ -18,7 +18,7 @@
 
 namespace cpp2sky {
 
-TracerImpl::TracerImpl(TracerConfig& config,
+TracerImpl::TracerImpl(const TracerConfig& config,
                        std::shared_ptr<grpc::ChannelCredentials> cred,
                        GrpcAsyncSegmentReporterStreamFactory& factory)
     : th_([this] { this->run(); }) {
@@ -37,7 +37,7 @@ TracerImpl::~TracerImpl() {
 }
 
 void TracerImpl::sendSegment(SegmentContextPtr obj) {
-  if (!obj) {
+  if (!obj || !obj->readyToSend()) {
     return;
   }
   client_->sendMessage(obj->createSegmentObject());
@@ -55,17 +55,16 @@ void TracerImpl::run() {
     TaggedStream* t_stream = deTag(got_tag);
     if (!ok) {
       client_->resetStream();
-      continue;
-    }
-    if (!t_stream->stream->handleOperation(t_stream->operation)) {
       client_->startStream();
+      continue;
     }
+    t_stream->stream->handleOperation(t_stream->operation);
   }
 }
 
-TracerPtr createInsecureGrpcTracer(TracerConfig& cfg) {
+TracerPtr createInsecureGrpcTracer(const TracerConfig& cfg) {
   return std::make_unique<TracerImpl>(cfg, grpc::InsecureChannelCredentials(),
                                       stream_factory);
 }
 
-}  // namespace cpp2sky
+}  // namespace cpp2sky

source/tracer_impl.h

@@ -26,7 +26,7 @@ using TracerResponseType = Commands;
 
 class TracerImpl : public Tracer {
  public:
-  TracerImpl(TracerConfig& config,
+  TracerImpl(const TracerConfig& config,
              std::shared_ptr<grpc::ChannelCredentials> cred,
              GrpcAsyncSegmentReporterStreamFactory& factory);
   ~TracerImpl();
@@ -43,6 +43,6 @@ class TracerImpl : public Tracer {
 
 static GrpcAsyncSegmentReporterStreamFactory stream_factory;
 
-TracerPtr createInsecureGrpcTracer(TracerConfig& cfg);
+TracerPtr createInsecureGrpcTracer(const TracerConfig& cfg);
 
 }  // namespace cpp2sky