forked from open-telemetry/opentelemetry-collector-contrib
-
Notifications
You must be signed in to change notification settings - Fork 0
/
helper.go
66 lines (53 loc) · 1.86 KB
/
helper.go
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
// Copyright The OpenTelemetry Authors
// SPDX-License-Identifier: Apache-2.0
package prometheusremotewriteexporter // import "github.com/open-telemetry/opentelemetry-collector-contrib/exporter/prometheusremotewriteexporter"
import (
"errors"
"sort"
"github.com/prometheus/prometheus/prompb"
)
// batchTimeSeries splits series into multiple batch write requests.
func batchTimeSeries(tsMap map[string]*prompb.TimeSeries, maxBatchByteSize int) ([]*prompb.WriteRequest, error) {
if len(tsMap) == 0 {
return nil, errors.New("invalid tsMap: cannot be empty map")
}
var requests []*prompb.WriteRequest
var tsArray []prompb.TimeSeries
sizeOfCurrentBatch := 0
for _, v := range tsMap {
sizeOfSeries := v.Size()
if sizeOfCurrentBatch+sizeOfSeries >= maxBatchByteSize {
wrapped := convertTimeseriesToRequest(tsArray)
requests = append(requests, wrapped)
tsArray = nil
sizeOfCurrentBatch = 0
}
tsArray = append(tsArray, *v)
sizeOfCurrentBatch += sizeOfSeries
}
if len(tsArray) != 0 {
wrapped := convertTimeseriesToRequest(tsArray)
requests = append(requests, wrapped)
}
return requests, nil
}
func convertTimeseriesToRequest(tsArray []prompb.TimeSeries) *prompb.WriteRequest {
// the remote_write endpoint only requires the timeseries.
// otlp defines it's own way to handle metric metadata
return &prompb.WriteRequest{
// Prometheus requires time series to be sorted by Timestamp to avoid out of order problems.
// See:
// * https://github.com/open-telemetry/wg-prometheus/issues/10
// * https://github.com/open-telemetry/opentelemetry-collector/issues/2315
Timeseries: orderBySampleTimestamp(tsArray),
}
}
func orderBySampleTimestamp(tsArray []prompb.TimeSeries) []prompb.TimeSeries {
for i := range tsArray {
sL := tsArray[i].Samples
sort.Slice(sL, func(i, j int) bool {
return sL[i].Timestamp < sL[j].Timestamp
})
}
return tsArray
}