|
13 | 13 | // limitations under the License. |
14 | 14 |
|
15 | 15 | #include "opencensus/stats/stats_exporter.h" |
| 16 | +#include "opencensus/stats/internal/stats_exporter_impl.h" |
16 | 17 |
|
17 | | -#include <iostream> |
18 | 18 | #include <thread> // NOLINT |
19 | 19 | #include <utility> |
20 | 20 | #include <vector> |
|
24 | 24 | #include "absl/time/clock.h" |
25 | 25 | #include "absl/time/time.h" |
26 | 26 | #include "opencensus/stats/internal/aggregation_window.h" |
| 27 | +#include "opencensus/stats/view_data.h" |
| 28 | +#include "opencensus/stats/view_descriptor.h" |
27 | 29 |
|
28 | 30 | namespace opencensus { |
29 | 31 | namespace stats { |
30 | 32 |
|
31 | | -class StatsExporterImpl { |
32 | | - public: |
33 | | - static StatsExporterImpl* Get() { |
34 | | - static StatsExporterImpl* global_stats_exporter_impl = |
35 | | - new StatsExporterImpl(); |
36 | | - return global_stats_exporter_impl; |
37 | | - } |
38 | | - |
39 | | - void AddView(const ViewDescriptor& view) { |
40 | | - absl::MutexLock l(&mu_); |
41 | | - views_[view.name()] = absl::make_unique<opencensus::stats::View>(view); |
42 | | - } |
43 | | - |
44 | | - void RemoveView(absl::string_view name) { |
45 | | - absl::MutexLock l(&mu_); |
46 | | - views_.erase(std::string(name)); |
47 | | - } |
| 33 | +// static |
| 34 | +StatsExporterImpl* StatsExporterImpl::Get() { |
| 35 | + static StatsExporterImpl* global_stats_exporter_impl = |
| 36 | + new StatsExporterImpl(); |
| 37 | + return global_stats_exporter_impl; |
| 38 | +} |
48 | 39 |
|
49 | | - // Adds a handler, which cannot be subsequently removed (except by |
50 | | - // ClearHandlersForTesting()). The background thread is started when the |
51 | | - // first handler is registered. |
52 | | - void RegisterPushHandler(std::unique_ptr<StatsExporter::Handler> handler) { |
53 | | - absl::MutexLock l(&mu_); |
54 | | - handlers_.push_back(std::move(handler)); |
55 | | - if (!thread_started_) { |
56 | | - StartExportThread(); |
57 | | - } |
58 | | - } |
| 40 | +void StatsExporterImpl::AddView(const ViewDescriptor& view) { |
| 41 | + absl::MutexLock l(&mu_); |
| 42 | + views_[view.name()] = absl::make_unique<opencensus::stats::View>(view); |
| 43 | +} |
59 | 44 |
|
60 | | - std::vector<std::pair<ViewDescriptor, ViewData>> GetViewData() { |
61 | | - absl::ReaderMutexLock l(&mu_); |
62 | | - std::vector<std::pair<ViewDescriptor, ViewData>> data; |
63 | | - data.reserve(views_.size()); |
64 | | - for (const auto& view : views_) { |
65 | | - data.emplace_back(view.second->descriptor(), view.second->GetData()); |
66 | | - } |
67 | | - return data; |
68 | | - } |
| 45 | +void StatsExporterImpl::RemoveView(absl::string_view name) { |
| 46 | + absl::MutexLock l(&mu_); |
| 47 | + views_.erase(std::string(name)); |
| 48 | +} |
69 | 49 |
|
70 | | - void Export() { |
71 | | - absl::ReaderMutexLock l(&mu_); |
72 | | - for (const auto& view : views_) { |
73 | | - SendToHandlers(view.second->descriptor(), view.second->GetData()); |
74 | | - } |
| 50 | +void StatsExporterImpl::RegisterPushHandler( |
| 51 | + std::unique_ptr<StatsExporter::Handler> handler) { |
| 52 | + absl::MutexLock l(&mu_); |
| 53 | + handlers_.push_back(std::move(handler)); |
| 54 | + if (!thread_started_) { |
| 55 | + StartExportThread(); |
75 | 56 | } |
| 57 | +} |
76 | 58 |
|
77 | | - void ClearHandlersForTesting() { |
78 | | - absl::MutexLock l(&mu_); |
79 | | - handlers_.clear(); |
| 59 | +std::vector<std::pair<ViewDescriptor, ViewData>> |
| 60 | +StatsExporterImpl::GetViewData() { |
| 61 | + absl::ReaderMutexLock l(&mu_); |
| 62 | + std::vector<std::pair<ViewDescriptor, ViewData>> data; |
| 63 | + data.reserve(views_.size()); |
| 64 | + for (const auto& view : views_) { |
| 65 | + data.emplace_back(view.second->descriptor(), view.second->GetData()); |
80 | 66 | } |
| 67 | + return data; |
| 68 | +} |
81 | 69 |
|
82 | | - private: |
83 | | - StatsExporterImpl() {} |
84 | | - |
85 | | - void SendToHandlers(const ViewDescriptor& descriptor, const ViewData& data) |
86 | | - SHARED_LOCKS_REQUIRED(mu_) { |
87 | | - for (auto& handler : handlers_) { |
88 | | - handler->ExportViewData(descriptor, data); |
89 | | - } |
| 70 | +void StatsExporterImpl::Export() { |
| 71 | + absl::ReaderMutexLock l(&mu_); |
| 72 | + for (const auto& view : views_) { |
| 73 | + SendToHandlers(view.second->descriptor(), view.second->GetData()); |
90 | 74 | } |
| 75 | +} |
91 | 76 |
|
92 | | - void StartExportThread() EXCLUSIVE_LOCKS_REQUIRED(mu_) { |
93 | | - t_ = std::thread(&StatsExporterImpl::RunWorkerLoop, this); |
94 | | - thread_started_ = true; |
95 | | - } |
| 77 | +void StatsExporterImpl::ClearHandlersForTesting() { |
| 78 | + absl::MutexLock l(&mu_); |
| 79 | + handlers_.clear(); |
| 80 | +} |
96 | 81 |
|
97 | | - // Loops forever, calling Export() every export_interval_. |
98 | | - void RunWorkerLoop() { |
99 | | - absl::Time next_export_time = absl::Now() + export_interval_; |
100 | | - while (true) { |
101 | | - // SleepFor() returns immediately when given a negative duration. |
102 | | - absl::SleepFor(next_export_time - absl::Now()); |
103 | | - // In case the last export took longer than the export interval, we |
104 | | - // calculate the next time from now. |
105 | | - next_export_time = absl::Now() + export_interval_; |
106 | | - Export(); |
107 | | - } |
| 82 | +void StatsExporterImpl::SendToHandlers(const ViewDescriptor& descriptor, |
| 83 | + const ViewData& data) |
| 84 | + SHARED_LOCKS_REQUIRED(mu_) { |
| 85 | + for (auto& handler : handlers_) { |
| 86 | + handler->ExportViewData(descriptor, data); |
108 | 87 | } |
| 88 | +} |
109 | 89 |
|
110 | | - const absl::Duration export_interval_ = absl::Seconds(10); |
111 | | - |
112 | | - mutable absl::Mutex mu_; |
113 | | - |
114 | | - std::vector<std::unique_ptr<StatsExporter::Handler>> handlers_ |
115 | | - GUARDED_BY(mu_); |
116 | | - std::unordered_map<std::string, std::unique_ptr<View>> views_ GUARDED_BY(mu_); |
117 | | - |
118 | | - bool thread_started_ GUARDED_BY(mu_) = false; |
119 | | - std::thread t_ GUARDED_BY(mu_); |
120 | | -}; |
| 90 | +void StatsExporterImpl::StartExportThread() EXCLUSIVE_LOCKS_REQUIRED(mu_) { |
| 91 | + t_ = std::thread(&StatsExporterImpl::RunWorkerLoop, this); |
| 92 | + thread_started_ = true; |
| 93 | +} |
121 | 94 |
|
122 | | -void StatsExporter::AddView(const ViewDescriptor& view) { |
123 | | - if (view.aggregation_window().type() == |
124 | | - AggregationWindow::Type::kCumulative) { |
125 | | - StatsExporterImpl::Get()->AddView(view); |
126 | | - } else { |
127 | | - std::cerr << "Only cumulative views may be registered for export.\n"; |
| 95 | +void StatsExporterImpl::RunWorkerLoop() { |
| 96 | + absl::Time next_export_time = absl::Now() + export_interval_; |
| 97 | + while (true) { |
| 98 | + // SleepFor() returns immediately when given a negative duration. |
| 99 | + absl::SleepFor(next_export_time - absl::Now()); |
| 100 | + // In case the last export took longer than the export interval, we |
| 101 | + // calculate the next time from now. |
| 102 | + next_export_time = absl::Now() + export_interval_; |
| 103 | + Export(); |
128 | 104 | } |
129 | 105 | } |
130 | 106 |
|
|
0 commit comments