-
Notifications
You must be signed in to change notification settings - Fork 5
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
agents: implement SpanCollector helper class
SpanCollector is a class that allows to start collecting spans and report them to a callback function running in a specific uv_loop_t thread. The collector can be configured with a minimum span count and a time interval to report the spans. In addition, the collector can be configured with a transform function that will be applied to the spans before calling the final callback. Use this class to dry the code in the `ZmqAgent` and `OTLPAgent`. PR-URL: #160 Reviewed-By: Trevor Norris <trev.norris@gmail.com>
- Loading branch information
1 parent
fda0c4e
commit 8c53110
Showing
9 changed files
with
249 additions
and
115 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,89 @@ | ||
| #include "span_collector.h" | ||
| #include "asserts-cpp/asserts.h" | ||
|
|
||
| namespace node { | ||
| namespace nsolid { | ||
|
|
||
| SpanCollector::SpanCollector(uv_loop_t* loop, | ||
| uint32_t trace_flags, | ||
| size_t min_span_count, | ||
| uint64_t interval): | ||
| loop_(loop), | ||
| trace_flags_(trace_flags), | ||
| min_span_count_(min_span_count), | ||
| span_msg_(new nsuv::ns_async()), | ||
| interval_(interval), | ||
| span_timer_(new nsuv::ns_timer()) { | ||
| } | ||
|
|
||
| SpanCollector::~SpanCollector() { | ||
| span_timer_->close_and_delete(); | ||
| span_msg_->close_and_delete(); | ||
| } | ||
|
|
||
| void SpanCollector::do_collect() { | ||
| // Create the instance with a callback that enqueues spans and sends a message | ||
| Tracer* tracer = Tracer::CreateInstance( | ||
| trace_flags_, | ||
| +[](Tracer*, Tracer::SpanStor span, WeakSpanCollector collector_wp) { | ||
| SharedSpanCollector collector = collector_wp.lock(); | ||
| if (collector == nullptr) { | ||
| return; | ||
| } | ||
|
|
||
| // Only notify the loop_ thread if the queue reaches min_span_count_to | ||
| // avoid too many uv_async_t::send() calls. | ||
| if (collector->span_msg_q_.enqueue(span) > collector->min_span_count_) { | ||
| ASSERT_EQ(0, collector->span_msg_->send()); | ||
| } | ||
| }, | ||
| weak_from_this()); | ||
| ASSERT_NOT_NULL(tracer); | ||
| tracer_.reset(tracer); | ||
|
|
||
| int er = span_msg_->init( | ||
| loop_, | ||
| +[](nsuv::ns_async*, WeakSpanCollector collector_wp) { | ||
| SharedSpanCollector collector = collector_wp.lock(); | ||
| if (collector == nullptr) { | ||
| return; | ||
| } | ||
|
|
||
| collector->process_spans(); | ||
| }, | ||
| weak_from_this()); | ||
| ASSERT_EQ(0, er); | ||
| er = span_timer_->init(loop_); | ||
| ASSERT_EQ(0, er); | ||
| er = span_timer_->start(+[](nsuv::ns_timer*, | ||
| WeakSpanCollector collector_wp) { | ||
| SharedSpanCollector collector = collector_wp.lock(); | ||
| if (collector == nullptr) { | ||
| return; | ||
| } | ||
|
|
||
| collector->process_spans(); | ||
| }, 0, interval_, weak_from_this()); | ||
| ASSERT_EQ(0, er); | ||
| } | ||
|
|
||
| void SpanCollector::process_spans() { | ||
| Tracer::SpanStor span; | ||
| SpanVector spans; | ||
| while (span_msg_q_.dequeue(span)) { | ||
| spans.push_back(std::move(span)); | ||
| } | ||
|
|
||
| // After processing, call the stored callback with the spans and additional | ||
| // data. | ||
| if (spans.size()) { | ||
| if (transform_callback_) { | ||
| transform_callback_(spans); | ||
| } else { | ||
| callback_(spans); | ||
| } | ||
| } | ||
| } | ||
|
|
||
| } // namespace nsolid | ||
| } // namespace node |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,91 @@ | ||
| #ifndef AGENTS_SRC_SPAN_COLLECTOR_H_ | ||
| #define AGENTS_SRC_SPAN_COLLECTOR_H_ | ||
|
|
||
| #include <nsolid.h> | ||
| #include <nsolid/thread_safe.h> | ||
| #include "nsuv-inl.h" | ||
| #include <vector> | ||
|
|
||
| namespace node { | ||
| namespace nsolid { | ||
|
|
||
| class SpanCollector; | ||
|
|
||
| using SharedSpanCollector = std::shared_ptr<SpanCollector>; | ||
| using WeakSpanCollector = std::weak_ptr<SpanCollector>; | ||
| using SpanVector = std::vector<Tracer::SpanStor>; | ||
|
|
||
| /* | ||
| * SpanCollector is a class that allows to start collecting spans and report | ||
| * them to a callback function running in a specific uv_loop_t thread. The | ||
| * collector can be configured with a minimum span count and a time interval to | ||
| * report the spans. In addition, the collector can be configured with a | ||
| * transform function that will be applied to the spans before calling the final | ||
| * callback. | ||
| */ | ||
| class SpanCollector: public std::enable_shared_from_this<SpanCollector> { | ||
| public: | ||
| explicit SpanCollector(uv_loop_t* loop, | ||
| uint32_t trace_flags, | ||
| size_t min_span_count, | ||
| uint64_t interval); | ||
| ~SpanCollector(); | ||
|
|
||
| template <typename Cb, typename... Data> | ||
| void Collect(Cb&& cb, Data&&... data) { | ||
| // Store the callback and data | ||
| callback_ = std::bind(std::forward<Cb>(cb), | ||
| std::placeholders::_1, | ||
| std::forward<Data>(data)...); | ||
| this->do_collect(); | ||
| } | ||
|
|
||
| template <typename Transform, typename Cb, typename... Data> | ||
| void CollectAndTransform(Transform&& transform, Cb&& cb, Data&&... data) { | ||
| // Store the transform function and callback | ||
| auto data_tuple = std::make_tuple(std::forward<Data>(data)...); | ||
| transform_callback_ = [transform = std::forward<Transform>(transform), | ||
| cb = std::forward<Cb>(cb), | ||
| data_tuple] (const SpanVector& spans) { | ||
| // Create a vector for transformed spans | ||
| using TransformedType = | ||
| decltype(transform(std::declval<Tracer::SpanStor>(), | ||
| std::declval<Data>()...)); | ||
| std::vector<TransformedType> transformed_spans; | ||
| transformed_spans.reserve(spans.size()); | ||
|
|
||
| // Apply the transform to each span | ||
| std::apply([&transform, &spans, &transformed_spans](const Data&... data) { | ||
| for (const auto& span : spans) { | ||
| transformed_spans.push_back(transform(span, data...)); | ||
| } | ||
| }, data_tuple); | ||
|
|
||
| // Call the callback with the transformed spans and additional data | ||
| std::apply([&cb, &transformed_spans](const Data&... data) { | ||
| cb(transformed_spans, data...); | ||
| }, data_tuple); | ||
| }; | ||
| this->do_collect(); | ||
| } | ||
|
|
||
| private: | ||
| void do_collect(); | ||
| void process_spans(); | ||
|
|
||
| uv_loop_t* loop_; | ||
| uint32_t trace_flags_; | ||
| size_t min_span_count_; | ||
| std::unique_ptr<Tracer> tracer_; | ||
| nsuv::ns_async* span_msg_; | ||
| TSQueue<Tracer::SpanStor> span_msg_q_; | ||
| uint64_t interval_; | ||
| nsuv::ns_timer* span_timer_; | ||
| std::function<void(const SpanVector&)> callback_ = nullptr; | ||
| std::function<void(const SpanVector&)> transform_callback_ = nullptr; | ||
| }; | ||
|
|
||
| } // namespace nsolid | ||
| } // namespace node | ||
|
|
||
| #endif // AGENTS_SRC_SPAN_COLLECTOR_H_ |
Oops, something went wrong.