blob: 44ea717a0bfa41c655274707d3b1fe2f263242b4 (
plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
|
// Copyright The OpenTelemetry Authors
// SPDX-License-Identifier: Apache-2.0
#pragma once
#ifdef ENABLE_METRICS_PREVIEW
# include "opentelemetry/_metrics/instrument.h"
# include "opentelemetry/sdk/_metrics/aggregator/aggregator.h"
# include "opentelemetry/version.h"
# include <cmath>
# include <memory>
# include <mutex>
# include <vector>
OPENTELEMETRY_BEGIN_NAMESPACE
namespace sdk
{
namespace metrics
{
/**
* This aggregator has two modes. In-order and quantile estimation.
*
* The first mode simply stores all values sent to the Update()
* function in a vector and maintains the order they were sent in.
*
* The second mode also stores all values sent to the Update()
* function in a vector but sorts this vector when Checkpoint()
* is called. This mode also includes a function, Quantile(),
* that estimates the quantiles of the recorded data.
*
* @tparam T the type of values stored in this aggregator.
*/
template <class T>
class ExactAggregator : public Aggregator<T>
{
public:
ExactAggregator(opentelemetry::metrics::InstrumentKind kind, bool quant_estimation = false)
{
static_assert(std::is_arithmetic<T>::value, "Not an arithmetic type");
this->kind_ = kind;
this->checkpoint_ = this->values_;
this->agg_kind_ = AggregatorKind::Exact;
quant_estimation_ = quant_estimation;
}
~ExactAggregator() = default;
ExactAggregator(const ExactAggregator &cp)
{
this->values_ = cp.values_;
this->checkpoint_ = cp.checkpoint_;
this->kind_ = cp.kind_;
this->agg_kind_ = cp.agg_kind_;
quant_estimation_ = cp.quant_estimation_;
// use default initialized mutex as they cannot be copied
}
/**
* Receives a captured value from the instrument and adds it to the values_ vector.
*
* @param val, the raw value used in aggregation
*/
void update(T val) override
{
this->mu_.lock();
this->updated_ = true;
this->values_.push_back(val);
this->mu_.unlock();
}
/**
* Checkpoints the current values. This function will overwrite the current checkpoint with the
* current value. Sorts the values_ vector if quant_estimation_ == true
*
*/
void checkpoint() override
{
this->mu_.lock();
this->updated_ = false;
if (quant_estimation_)
{
std::sort(this->values_.begin(), this->values_.end());
}
this->checkpoint_ = this->values_;
this->values_.clear();
this->mu_.unlock();
}
/**
* Merges two exact aggregators' values_ vectors together.
*
* @param other the aggregator to merge with this aggregator
*/
void merge(const ExactAggregator &other)
{
if (this->kind_ == other.kind_)
{
this->mu_.lock();
// First merge values
this->values_.insert(this->values_.end(), other.values_.begin(), other.values_.end());
// Now merge checkpoints
this->checkpoint_.insert(this->checkpoint_.end(), other.checkpoint_.begin(),
other.checkpoint_.end());
this->mu_.unlock();
}
else
{
// Log error
return;
}
}
/**
* Performs quantile estimation on the checkpoint vector in this aggregator.
* This function only works if quant_estimation_ == true.
* @param q the quantile to estimate. 0 <= q <= 1
* @return the nearest value in the vector to the exact quantile.
*/
T get_quantiles(double q) override
{
if (!quant_estimation_)
{
// Log error
# if __EXCEPTIONS
throw std::domain_error("Exact aggregator is not in quantile estimation mode!");
# else
std::terminate();
# endif
}
if (this->checkpoint_.size() == 0 || q < 0 || q > 1)
{
// Log error
# if __EXCEPTIONS
throw std::invalid_argument("Arg 'q' must be between 0 and 1, inclusive");
# else
std::terminate();
# endif
}
else if (q == 0 || this->checkpoint_.size() == 1)
{
return this->checkpoint_[0];
}
else if (q == 1)
{
return this->checkpoint_[this->checkpoint_.size() - 1];
}
else
{
float position = float(float(this->checkpoint_.size() - 1) * q);
int ceiling = int(ceil(position));
return this->checkpoint_[ceiling];
}
}
//////////////////////////ACCESSOR FUNCTIONS//////////////////////////
std::vector<T> get_checkpoint() override { return this->checkpoint_; }
std::vector<T> get_values() override { return this->values_; }
bool get_quant_estimation() override { return quant_estimation_; }
private:
bool quant_estimation_; // Used to switch between in-order and quantile estimation modes
};
} // namespace metrics
} // namespace sdk
OPENTELEMETRY_END_NAMESPACE
#endif
|
