Codebase list golang-github-go-kit-kit / 7dd0815
use batch values API for CloudWatch PutMetric data call (#960) * use batch values API for CloudWatch PutMetric data call which was introduced at https://github.com/aws/aws-sdk-go/blob/master/CHANGELOG.md#release-v11536-2018-09-17 * fix test, so they can accept the list of received values from the gauge * use batch api always Taras authored 3 years ago GitHub committed 3 years ago
9 changed file(s) with 106 addition(s) and 47 deletion(s). Raw diff Collapse all Expand all
+24
-4
metrics/cloudwatch/cloudwatch.go less more
1919
2020 const (
2121 maxConcurrentRequests = 20
22 maxValuesInABatch = 150
2223 )
2324
2425 type Percentiles []struct {
173174 })
174175
175176 cw.gauges.Reset().Walk(func(name string, lvs lv.LabelValues, values []float64) bool {
176 value := last(values)
177 datums = append(datums, &cloudwatch.MetricDatum{
177 datum := &cloudwatch.MetricDatum{
178178 MetricName: aws.String(name),
179179 Dimensions: makeDimensions(lvs...),
180 Value: aws.Float64(value),
181180 Timestamp: aws.Time(now),
182 })
181 }
182
183 if len(values) == 0 {
184 return true
185 }
186
187 // CloudWatch Put Metrics API (https://docs.aws.amazon.com/AmazonCloudWatch/latest/APIReference/API_MetricDatum.html)
188 // expects batch of unique values including the array of corresponding counts
189 valuesCounter := make(map[float64]int)
190 for _, v := range values {
191 valuesCounter[v]++
192 }
193
194 for value, count := range valuesCounter {
195 if len(datum.Values) == maxValuesInABatch {
196 break
197 }
198 datum.Values = append(datum.Values, aws.Float64(value))
199 datum.Counts = append(datum.Counts, aws.Float64(float64(count)))
200 }
201
202 datums = append(datums, datum)
183203 return true
184204 })
185205
+66
-28
metrics/cloudwatch/cloudwatch_test.go less more
1717 type mockCloudWatch struct {
1818 cloudwatchiface.CloudWatchAPI
1919 mtx sync.RWMutex
20 valuesReceived map[string]float64
20 valuesReceived map[string][]float64
2121 dimensionsReceived map[string][]*cloudwatch.Dimension
2222 }
2323
2424 func newMockCloudWatch() *mockCloudWatch {
2525 return &mockCloudWatch{
26 valuesReceived: map[string]float64{},
26 valuesReceived: map[string][]float64{},
2727 dimensionsReceived: map[string][]*cloudwatch.Dimension{},
2828 }
2929 }
3232 mcw.mtx.Lock()
3333 defer mcw.mtx.Unlock()
3434 for _, datum := range input.MetricData {
35 mcw.valuesReceived[*datum.MetricName] = *datum.Value
35 if len(datum.Values) > 0 {
36 for _, v := range datum.Values {
37 mcw.valuesReceived[*datum.MetricName] = append(mcw.valuesReceived[*datum.MetricName], *v)
38 }
39 } else {
40 mcw.valuesReceived[*datum.MetricName] = append(mcw.valuesReceived[*datum.MetricName], *datum.Value)
41 }
3642 mcw.dimensionsReceived[*datum.MetricName] = datum.Dimensions
3743 }
3844 return nil, nil
7581 cw := New(namespace, svc, WithLogger(log.NewNopLogger()))
7682 counter := cw.NewCounter(name).With(label, value)
7783 valuef := func() float64 {
78 err := cw.Send()
79 if err != nil {
80 t.Fatal(err)
81 }
82 svc.mtx.RLock()
83 defer svc.mtx.RUnlock()
84 return svc.valuesReceived[name]
84 if err := cw.Send(); err != nil {
85 t.Fatal(err)
86 }
87 svc.mtx.RLock()
88 defer svc.mtx.RUnlock()
89 value := svc.valuesReceived[name][len(svc.valuesReceived[name])-1]
90 delete(svc.valuesReceived, name)
91
92 return value
8593 }
8694 if err := teststat.TestCounter(counter, valuef); err != nil {
8795 t.Fatal(err)
122130 }
123131
124132 for i, name := range names {
125 if svc.valuesReceived[name] != wants[i] {
133 if l := len(svc.valuesReceived[name]); l == 0 && wants[i] == 0 {
134 continue
135 } else if l != 1 {
136 t.Fatalf("one value expected, got %d", l)
137 }
138
139 if svc.valuesReceived[name][0] != wants[i] {
126140 t.Fatalf("want %f, have %f", wants[i], svc.valuesReceived[name])
127141 }
128142 if err := svc.testDimensions(name, labels[i], values[i]); err != nil {
137151 svc := newMockCloudWatch()
138152 cw := New(namespace, svc, WithLogger(log.NewNopLogger()))
139153 gauge := cw.NewGauge(name).With(label, value)
140 valuef := func() float64 {
141 err := cw.Send()
142 if err != nil {
143 t.Fatal(err)
144 }
145 svc.mtx.RLock()
146 defer svc.mtx.RUnlock()
147 return svc.valuesReceived[name]
148 }
154 valuef := func() []float64 {
155 if err := cw.Send(); err != nil {
156 t.Fatal(err)
157 }
158 svc.mtx.RLock()
159 res := svc.valuesReceived[name]
160 delete(svc.valuesReceived, name)
161 defer svc.mtx.RUnlock()
162 return res
163 }
164
149165 if err := teststat.TestGauge(gauge, valuef); err != nil {
150166 t.Fatal(err)
151167 }
169185 if err != nil {
170186 t.Fatal(err)
171187 }
172 svc.mtx.RLock()
173 defer svc.mtx.RUnlock()
174 p50 = svc.valuesReceived[n50]
175 p90 = svc.valuesReceived[n90]
176 p95 = svc.valuesReceived[n95]
177 p99 = svc.valuesReceived[n99]
188
189 svc.mtx.RLock()
190 defer svc.mtx.RUnlock()
191 if len(svc.valuesReceived[n50]) > 0 {
192 p50 = svc.valuesReceived[n50][0]
193 delete(svc.valuesReceived, n50)
194 }
195
196 if len(svc.valuesReceived[n90]) > 0 {
197 p90 = svc.valuesReceived[n90][0]
198 delete(svc.valuesReceived, n90)
199 }
200
201 if len(svc.valuesReceived[n95]) > 0 {
202 p95 = svc.valuesReceived[n95][0]
203 delete(svc.valuesReceived, n95)
204 }
205
206 if len(svc.valuesReceived[n99]) > 0 {
207 p99 = svc.valuesReceived[n99][0]
208 delete(svc.valuesReceived, n99)
209 }
178210 return
179211 }
180212 if err := teststat.TestHistogram(histogram, quantiles, 0.01); err != nil {
206238 }
207239 svc.mtx.RLock()
208240 defer svc.mtx.RUnlock()
209 p50 = svc.valuesReceived[n50]
210 p90 = svc.valuesReceived[n90]
241 if len(svc.valuesReceived[n50]) > 0 {
242 p50 = svc.valuesReceived[n50][0]
243 delete(svc.valuesReceived, n50)
244 }
245 if len(svc.valuesReceived[n90]) > 0 {
246 p90 = svc.valuesReceived[n90][0]
247 delete(svc.valuesReceived, n90)
248 }
211249
212250 // our teststat.TestHistogram wants us to give p95 and p99,
213251 // but with custom percentiles we don't have those.
+1
-1
metrics/expvar/expvar_test.go less more
1616
1717 func TestGauge(t *testing.T) {
1818 gauge := NewGauge("expvar_gauge").With("label values", "not supported").(*Gauge)
19 value := func() float64 { f, _ := strconv.ParseFloat(gauge.f.String(), 64); return f }
19 value := func() []float64 { f, _ := strconv.ParseFloat(gauge.f.String(), 64); return []float64{f} }
2020 if err := teststat.TestGauge(gauge, value); err != nil {
2121 t.Fatal(err)
2222 }
+1
-1
metrics/generic/generic_test.go less more
4444 if want, have := name, gauge.Name; want != have {
4545 t.Errorf("Name: want %q, have %q", want, have)
4646 }
47 value := gauge.Value
47 value := func() []float64 { return []float64{gauge.Value()} }
4848 if err := teststat.TestGauge(gauge, value); err != nil {
4949 t.Fatal(err)
5050 }
+2
-2
metrics/influx/influx_test.go less more
3333 in := New(map[string]string{"foo": "alpha"}, influxdb.BatchPointsConfig{}, log.NewNopLogger())
3434 re := regexp.MustCompile(`influx_gauge,foo=alpha value=([0-9\.]+) [0-9]+`)
3535 gauge := in.NewGauge("influx_gauge")
36 value := func() float64 {
36 value := func() []float64 {
3737 client := &bufWriter{}
3838 in.WriteTo(client)
3939 match := re.FindStringSubmatch(client.buf.String())
4040 f, _ := strconv.ParseFloat(match[1], 64)
41 return f
41 return []float64{f}
4242 }
4343 if err := teststat.TestGauge(gauge, value); err != nil {
4444 t.Fatal(err)
+1
-1
metrics/pcp/pcp_test.go less more
3939
4040 gauge = gauge.With("label values", "not supported").(*Gauge)
4141
42 value := func() float64 { f := gauge.g.Val(); return f }
42 value := func() []float64 { f := gauge.g.Val(); return []float64{f} }
4343 if err := teststat.TestGauge(gauge, value); err != nil {
4444 t.Fatal(err)
4545 }
+2
-2
metrics/prometheus/prometheus_test.go less more
6767 Help: "This is a different help string.",
6868 }, []string{"foo"}).With("foo", "bar")
6969
70 value := func() float64 {
70 value := func() []float64 {
7171 matches := re.FindStringSubmatch(scrape())
7272 f, _ := strconv.ParseFloat(matches[1], 64)
73 return f
73 return []float64{f}
7474 }
7575
7676 if err := teststat.TestGauge(gauge, value); err != nil {
+3
-3
metrics/teststat/buffers.go less more
2222 // LastLine expects a regex whose first capture group can be parsed as a
2323 // float64. It will dump the WriterTo and parse each line, expecting to find a
2424 // match. It returns the final captured float.
25 func LastLine(w io.WriterTo, regex string) func() float64 {
26 return func() float64 {
25 func LastLine(w io.WriterTo, regex string) func() []float64 {
26 return func() []float64 {
2727 _, final := stats(w, regex, nil)
28 return final
28 return []float64{final}
2929 }
3030 }
3131
+6
-5
metrics/teststat/teststat.go less more
55 "fmt"
66 "math"
77 "math/rand"
8 "reflect"
89 "strings"
910
1011 "github.com/go-kit/kit/metrics"
3738
3839 // TestGauge puts some values through the gauge, and then calls the value func
3940 // to check that the gauge has the correct final value.
40 func TestGauge(gauge metrics.Gauge, value func() float64) error {
41 func TestGauge(gauge metrics.Gauge, value func() []float64) error {
4142 a := rand.Perm(100)
4243 n := rand.Intn(len(a))
4344
44 var want float64
45 var want []float64
4546 for i := 0; i < n; i++ {
4647 f := float64(a[i])
4748 gauge.Set(f)
48 want = f
49 want = append(want, f)
4950 }
5051
5152 for i := 0; i < n; i++ {
5253 f := float64(a[i])
5354 gauge.Add(f)
54 want += f
55 want[len(want)-1] += f
5556 }
5657
57 if have := value(); want != have {
58 if have := value(); reflect.DeepEqual(want, have) {
5859 return fmt.Errorf("want %f, have %f", want, have)
5960 }
6061