ceph/src/jaegertracing/opentelemetry-cpp/exporters/jaeger/src/thrift_sender.cc

   1 // Copyright The OpenTelemetry Authors
   2 // SPDX-License-Identifier: Apache-2.0
   3
   4 #include "thrift_sender.h"
   5 #include <opentelemetry/exporters/jaeger/recordable.h>
   6 #include "opentelemetry/sdk/common/global_log_handler.h"
   7 #include "udp_transport.h"
   8
   9 OPENTELEMETRY_BEGIN_NAMESPACE
  10 namespace exporter
  11 {
  12 namespace jaeger
  13 {
  14
  15 using namespace jaegertracing;
  16
  17 ThriftSender::ThriftSender(std::unique_ptr<Transport> &&transport)
  18     : transport_(std::move(transport)),
  19       protocol_factory_(new apache::thrift::protocol::TCompactProtocolFactory()),
  20       thrift_buffer_(new apache::thrift::transport::TMemoryBuffer(transport_->MaxPacketSize()))
  21 {}
  22
  23 int ThriftSender::Append(std::unique_ptr<JaegerRecordable> &&span) noexcept
  24 {
  25   if (span == nullptr)
  26   {
  27     return 0;
  28   }
  29
  30   uint32_t max_span_bytes = transport_->MaxPacketSize() - kEmitBatchOverhead;
  31   if (process_.serviceName.empty())
  32   {
  33     process_.serviceName = span->ServiceName();
  34     process_.__set_tags(span->ResourceTags());
  35
  36     process_bytes_size_ = CalcSizeOfSerializedThrift(process_);
  37     max_span_bytes -= process_bytes_size_;
  38   }
  39
  40   auto jaeger_span = std::unique_ptr<thrift::Span>(span->Span());
  41   jaeger_span->__set_tags(span->Tags());
  42   jaeger_span->__set_logs(span->Logs());
  43   jaeger_span->__set_references(span->References());
  44
  45   const uint32_t span_size = CalcSizeOfSerializedThrift(*jaeger_span);
  46   if (span_size > max_span_bytes)
  47   {
  48     OTEL_INTERNAL_LOG_ERROR("[JAEGER TRACE Exporter] Append() failed: too large span");
  49     return 0;
  50   }
  51
  52   byte_buffer_size_ += span_size;
  53   if (byte_buffer_size_ <= max_span_bytes)
  54   {
  55     span_buffer_.push_back(*jaeger_span);
  56     if (byte_buffer_size_ < max_span_bytes)
  57     {
  58       return 0;
  59     }
  60     else
  61     {
  62       // byte buffer is full so flush it before appending new span.
  63       return Flush();
  64     }
  65   }
  66
  67   const auto flushed = Flush();
  68   span_buffer_.push_back(*jaeger_span);
  69   byte_buffer_size_ = span_size + process_bytes_size_;
  70
  71   return flushed;
  72 }
  73
  74 int ThriftSender::Flush()
  75 {
  76   if (span_buffer_.empty())
  77   {
  78     return 0;
  79   }
  80
  81   thrift::Batch batch;
  82   batch.__set_process(process_);
  83   batch.__set_spans(span_buffer_);
  84
  85   int spans_flushed = transport_->EmitBatch(batch);
  86
  87   ResetBuffers();
  88
  89   return spans_flushed;
  90 }
  91
  92 void ThriftSender::Close()
  93 {
  94   Flush();
  95 }
  96
  97 }  // namespace jaeger
  98 }  // namespace exporter
  99 OPENTELEMETRY_END_NAMESPACE