Skip to content

Commit d9eb17f

Browse files
shbiengshbieng
authored
Merge pull request #808 from jmacd/jmacd/agg_refactor
Add aggregation.Kind and rename sdk/export/metric/aggregator to aggregation
2 parents b3133c4 + 8fea00b commit d9eb17f

File tree

26 files changed

+249
-138
lines changed

26 files changed

+249
-138
lines changed

‎exporters/metric/prometheus/prometheus.go

+9-9
Original file line numberDiff line numberDiff line change
@@ -27,7 +27,7 @@ import (
2727
"go.opentelemetry.io/otel/api/label"
2828
"go.opentelemetry.io/otel/api/metric"
2929
export "go.opentelemetry.io/otel/sdk/export/metric"
30-
"go.opentelemetry.io/otel/sdk/export/metric/aggregator"
30+
"go.opentelemetry.io/otel/sdk/export/metric/aggregation"
3131
"go.opentelemetry.io/otel/sdk/metric/controller/pull"
3232
"go.opentelemetry.io/otel/sdk/metric/selector/simple"
3333
)
@@ -216,11 +216,11 @@ func (c *collector) Collect(ch chan<- prometheus.Metric) {
216216

217217
desc := c.toDesc(record, labelKeys)
218218

219-
if hist, ok := agg.(aggregator.Histogram); ok {
219+
if hist, ok := agg.(aggregation.Histogram); ok {
220220
if err := c.exportHistogram(ch, hist, numberKind, desc, labels); err != nil {
221221
return fmt.Errorf("exporting histogram: %w", err)
222222
}
223-
} else if dist, ok := agg.(aggregator.Distribution); ok {
223+
} else if dist, ok := agg.(aggregation.Distribution); ok {
224224
// TODO: summaries values are never being resetted.
225225
// As measurements are recorded, new records starts to have less impact on these summaries.
226226
// We should implement an solution that is similar to the Prometheus Clients
@@ -232,11 +232,11 @@ func (c *collector) Collect(ch chan<- prometheus.Metric) {
232232
if err := c.exportSummary(ch, dist, numberKind, desc, labels); err != nil {
233233
return fmt.Errorf("exporting summary: %w", err)
234234
}
235-
} else if sum, ok := agg.(aggregator.Sum); ok {
235+
} else if sum, ok := agg.(aggregation.Sum); ok {
236236
if err := c.exportCounter(ch, sum, numberKind, desc, labels); err != nil {
237237
return fmt.Errorf("exporting counter: %w", err)
238238
}
239-
} else if lastValue, ok := agg.(aggregator.LastValue); ok {
239+
} else if lastValue, ok := agg.(aggregation.LastValue); ok {
240240
if err := c.exportLastValue(ch, lastValue, numberKind, desc, labels); err != nil {
241241
return fmt.Errorf("exporting last value: %w", err)
242242
}
@@ -248,7 +248,7 @@ func (c *collector) Collect(ch chan<- prometheus.Metric) {
248248
}
249249
}
250250

251-
func (c *collector) exportLastValue(ch chan<- prometheus.Metric, lvagg aggregator.LastValue, kind metric.NumberKind, desc *prometheus.Desc, labels []string) error {
251+
func (c *collector) exportLastValue(ch chan<- prometheus.Metric, lvagg aggregation.LastValue, kind metric.NumberKind, desc *prometheus.Desc, labels []string) error {
252252
lv, _, err := lvagg.LastValue()
253253
if err != nil {
254254
return fmt.Errorf("error retrieving last value: %w", err)
@@ -263,7 +263,7 @@ func (c *collector) exportLastValue(ch chan<- prometheus.Metric, lvagg aggregato
263263
return nil
264264
}
265265

266-
func (c *collector) exportCounter(ch chan<- prometheus.Metric, sum aggregator.Sum, kind metric.NumberKind, desc *prometheus.Desc, labels []string) error {
266+
func (c *collector) exportCounter(ch chan<- prometheus.Metric, sum aggregation.Sum, kind metric.NumberKind, desc *prometheus.Desc, labels []string) error {
267267
v, err := sum.Sum()
268268
if err != nil {
269269
return fmt.Errorf("error retrieving counter: %w", err)
@@ -278,7 +278,7 @@ func (c *collector) exportCounter(ch chan<- prometheus.Metric, sum aggregator.Su
278278
return nil
279279
}
280280

281-
func (c *collector) exportSummary(ch chan<- prometheus.Metric, dist aggregator.Distribution, kind metric.NumberKind, desc *prometheus.Desc, labels []string) error {
281+
func (c *collector) exportSummary(ch chan<- prometheus.Metric, dist aggregation.Distribution, kind metric.NumberKind, desc *prometheus.Desc, labels []string) error {
282282
count, err := dist.Count()
283283
if err != nil {
284284
return fmt.Errorf("error retrieving count: %w", err)
@@ -305,7 +305,7 @@ func (c *collector) exportSummary(ch chan<- prometheus.Metric, dist aggregator.D
305305
return nil
306306
}
307307

308-
func (c *collector) exportHistogram(ch chan<- prometheus.Metric, hist aggregator.Histogram, kind metric.NumberKind, desc *prometheus.Desc, labels []string) error {
308+
func (c *collector) exportHistogram(ch chan<- prometheus.Metric, hist aggregation.Histogram, kind metric.NumberKind, desc *prometheus.Desc, labels []string) error {
309309
buckets, err := hist.Histogram()
310310
if err != nil {
311311
return fmt.Errorf("error retrieving histogram: %w", err)

‎exporters/metric/stdout/stdout.go

+6-6
Original file line numberDiff line numberDiff line change
@@ -27,7 +27,7 @@ import (
2727
"go.opentelemetry.io/otel/api/label"
2828

2929
export "go.opentelemetry.io/otel/sdk/export/metric"
30-
"go.opentelemetry.io/otel/sdk/export/metric/aggregator"
30+
"go.opentelemetry.io/otel/sdk/export/metric/aggregation"
3131
"go.opentelemetry.io/otel/sdk/metric/controller/push"
3232
"go.opentelemetry.io/otel/sdk/metric/selector/simple"
3333
)
@@ -98,7 +98,7 @@ func NewRawExporter(config Config) (*Exporter, error) {
9898
} else {
9999
for _, q := range config.Quantiles {
100100
if q < 0 || q > 1 {
101-
return nil, aggregator.ErrInvalidQuantile
101+
return nil, aggregation.ErrInvalidQuantile
102102
}
103103
}
104104
}
@@ -164,15 +164,15 @@ func (e *Exporter) Export(_ context.Context, checkpointSet export.CheckpointSet)
164164

165165
var expose expoLine
166166

167-
if sum, ok := agg.(aggregator.Sum); ok {
167+
if sum, ok := agg.(aggregation.Sum); ok {
168168
value, err := sum.Sum()
169169
if err != nil {
170170
return err
171171
}
172172
expose.Sum = value.AsInterface(kind)
173173
}
174174

175-
if mmsc, ok := agg.(aggregator.MinMaxSumCount); ok {
175+
if mmsc, ok := agg.(aggregation.MinMaxSumCount); ok {
176176
count, err := mmsc.Count()
177177
if err != nil {
178178
return err
@@ -191,7 +191,7 @@ func (e *Exporter) Export(_ context.Context, checkpointSet export.CheckpointSet)
191191
}
192192
expose.Min = min.AsInterface(kind)
193193

194-
if dist, ok := agg.(aggregator.Distribution); ok && len(e.config.Quantiles) != 0 {
194+
if dist, ok := agg.(aggregation.Distribution); ok && len(e.config.Quantiles) != 0 {
195195
summary := make([]expoQuantile, len(e.config.Quantiles))
196196
expose.Quantiles = summary
197197

@@ -208,7 +208,7 @@ func (e *Exporter) Export(_ context.Context, checkpointSet export.CheckpointSet)
208208
}
209209
}
210210
}
211-
} else if lv, ok := agg.(aggregator.LastValue); ok {
211+
} else if lv, ok := agg.(aggregation.LastValue); ok {
212212
value, timestamp, err := lv.LastValue()
213213
if err != nil {
214214
return err

‎exporters/metric/stdout/stdout_test.go

+2-2
Original file line numberDiff line numberDiff line change
@@ -29,7 +29,7 @@ import (
2929
"go.opentelemetry.io/otel/exporters/metric/stdout"
3030
"go.opentelemetry.io/otel/exporters/metric/test"
3131
export "go.opentelemetry.io/otel/sdk/export/metric"
32-
"go.opentelemetry.io/otel/sdk/export/metric/aggregator"
32+
"go.opentelemetry.io/otel/sdk/export/metric/aggregation"
3333
"go.opentelemetry.io/otel/sdk/metric/aggregator/array"
3434
"go.opentelemetry.io/otel/sdk/metric/aggregator/ddsketch"
3535
"go.opentelemetry.io/otel/sdk/metric/aggregator/lastvalue"
@@ -80,7 +80,7 @@ func TestStdoutInvalidQuantile(t *testing.T) {
8080
Quantiles: []float64{1.1, 0.9},
8181
})
8282
require.Error(t, err, "Invalid quantile error expected")
83-
require.Equal(t, aggregator.ErrInvalidQuantile, err)
83+
require.Equal(t, aggregation.ErrInvalidQuantile, err)
8484
}
8585

8686
func TestStdoutTimestamp(t *testing.T) {

‎exporters/metric/test/test.go

+2-2
Original file line numberDiff line numberDiff line change
@@ -23,7 +23,7 @@ import (
2323
"go.opentelemetry.io/otel/api/label"
2424
"go.opentelemetry.io/otel/api/metric"
2525
export "go.opentelemetry.io/otel/sdk/export/metric"
26-
"go.opentelemetry.io/otel/sdk/export/metric/aggregator"
26+
"go.opentelemetry.io/otel/sdk/export/metric/aggregation"
2727
"go.opentelemetry.io/otel/sdk/metric/aggregator/array"
2828
"go.opentelemetry.io/otel/sdk/metric/aggregator/histogram"
2929
"go.opentelemetry.io/otel/sdk/metric/aggregator/lastvalue"
@@ -117,7 +117,7 @@ func (p *CheckpointSet) updateAggregator(desc *metric.Descriptor, newAgg export.
117117

118118
func (p *CheckpointSet) ForEach(f func(export.Record) error) error {
119119
for _, r := range p.updates {
120-
if err := f(r); err != nil && !errors.Is(err, aggregator.ErrNoData) {
120+
if err := f(r); err != nil && !errors.Is(err, aggregation.ErrNoData) {
121121
return err
122122
}
123123
}

‎exporters/otlp/internal/transform/metric.go

+6-6
Original file line numberDiff line numberDiff line change
@@ -30,7 +30,7 @@ import (
3030
"go.opentelemetry.io/otel/api/label"
3131
"go.opentelemetry.io/otel/api/metric"
3232
export "go.opentelemetry.io/otel/sdk/export/metric"
33-
"go.opentelemetry.io/otel/sdk/export/metric/aggregator"
33+
"go.opentelemetry.io/otel/sdk/export/metric/aggregation"
3434
"go.opentelemetry.io/otel/sdk/resource"
3535
)
3636

@@ -230,17 +230,17 @@ func Record(r export.Record) (*metricpb.Metric, error) {
230230
d := r.Descriptor()
231231
l := r.Labels()
232232
switch a := r.Aggregator().(type) {
233-
case aggregator.MinMaxSumCount:
233+
case aggregation.MinMaxSumCount:
234234
return minMaxSumCount(d, l, a)
235-
case aggregator.Sum:
235+
case aggregation.Sum:
236236
return sum(d, l, a)
237237
default:
238238
return nil, fmt.Errorf("%w: %v", ErrUnimplementedAgg, a)
239239
}
240240
}
241241

242242
// sum transforms a Sum Aggregator into an OTLP Metric.
243-
func sum(desc *metric.Descriptor, labels *label.Set, a aggregator.Sum) (*metricpb.Metric, error) {
243+
func sum(desc *metric.Descriptor, labels *label.Set, a aggregation.Sum) (*metricpb.Metric, error) {
244244
sum, err := a.Sum()
245245
if err != nil {
246246
return nil, err
@@ -275,7 +275,7 @@ func sum(desc *metric.Descriptor, labels *label.Set, a aggregator.Sum) (*metricp
275275

276276
// minMaxSumCountValue returns the values of the MinMaxSumCount Aggregator
277277
// as discret values.
278-
func minMaxSumCountValues(a aggregator.MinMaxSumCount) (min, max, sum metric.Number, count int64, err error) {
278+
func minMaxSumCountValues(a aggregation.MinMaxSumCount) (min, max, sum metric.Number, count int64, err error) {
279279
if min, err = a.Min(); err != nil {
280280
return
281281
}
@@ -292,7 +292,7 @@ func minMaxSumCountValues(a aggregator.MinMaxSumCount) (min, max, sum metric.Num
292292
}
293293

294294
// minMaxSumCount transforms a MinMaxSumCount Aggregator into an OTLP Metric.
295-
func minMaxSumCount(desc *metric.Descriptor, labels *label.Set, a aggregator.MinMaxSumCount) (*metricpb.Metric, error) {
295+
func minMaxSumCount(desc *metric.Descriptor, labels *label.Set, a aggregation.MinMaxSumCount) (*metricpb.Metric, error) {
296296
min, max, sum, count, err := minMaxSumCountValues(a)
297297
if err != nil {
298298
return nil, err

‎exporters/otlp/internal/transform/metric_test.go

+3-3
Original file line numberDiff line numberDiff line change
@@ -27,7 +27,7 @@ import (
2727
"go.opentelemetry.io/otel/api/label"
2828
"go.opentelemetry.io/otel/api/metric"
2929
"go.opentelemetry.io/otel/api/unit"
30-
"go.opentelemetry.io/otel/sdk/export/metric/aggregator"
30+
"go.opentelemetry.io/otel/sdk/export/metric/aggregation"
3131
"go.opentelemetry.io/otel/sdk/metric/aggregator/minmaxsumcount"
3232
sumAgg "go.opentelemetry.io/otel/sdk/metric/aggregator/sum"
3333
)
@@ -86,7 +86,7 @@ func TestMinMaxSumCountValue(t *testing.T) {
8686

8787
// Prior to checkpointing ErrNoData should be returned.
8888
_, _, _, _, err := minMaxSumCountValues(mmsc)
89-
assert.EqualError(t, err, aggregator.ErrNoData.Error())
89+
assert.EqualError(t, err, aggregation.ErrNoData.Error())
9090

9191
// Checkpoint to set non-zero values
9292
mmsc.Checkpoint(&metric.Descriptor{})
@@ -198,7 +198,7 @@ func TestMinMaxSumCountPropagatesErrors(t *testing.T) {
198198
mmsc := minmaxsumcount.New(&metric.Descriptor{})
199199
_, _, _, _, err := minMaxSumCountValues(mmsc)
200200
assert.Error(t, err)
201-
assert.Equal(t, aggregator.ErrNoData, err)
201+
assert.Equal(t, aggregation.ErrNoData, err)
202202
}
203203

204204
func TestSumMetricDescriptor(t *testing.T) {

‎exporters/otlp/otlp_metric_test.go

+2-2
Original file line numberDiff line numberDiff line change
@@ -30,7 +30,7 @@ import (
3030
"go.opentelemetry.io/otel/api/label"
3131
"go.opentelemetry.io/otel/api/metric"
3232
metricsdk "go.opentelemetry.io/otel/sdk/export/metric"
33-
"go.opentelemetry.io/otel/sdk/export/metric/aggregator"
33+
"go.opentelemetry.io/otel/sdk/export/metric/aggregation"
3434
"go.opentelemetry.io/otel/sdk/metric/aggregator/minmaxsumcount"
3535
"go.opentelemetry.io/otel/sdk/metric/aggregator/sum"
3636
"go.opentelemetry.io/otel/sdk/resource"
@@ -67,7 +67,7 @@ type checkpointSet struct {
6767

6868
func (m *checkpointSet) ForEach(fn func(metricsdk.Record) error) error {
6969
for _, r := range m.records {
70-
if err := fn(r); err != nil && err != aggregator.ErrNoData {
70+
if err := fn(r); err != nil && err != aggregation.ErrNoData {
7171
return err
7272
}
7373
}

0 commit comments

Comments
 (0)