1
0
mirror of https://github.com/open-telemetry/opentelemetry-go.git synced 2024-12-12 10:04:29 +02:00
opentelemetry-go/sdk/metric/stress_test.go
Joshua MacDonald e8546e3bc5
Remove Labelset (#595)
* Remove LabelSet frmo api/metric

* SDK tests pass

* Restore benchmarks

* All tests pass

* Remove all mentions of LabelSet

* Test RecordBatch

* Batch test

* Improves benchmark (some)

* Move the benchmark to match HEAD

* Align labels for GOARCH=386

* Add alignment test

* Disable the stress test fo GOARCH=386

* Fix bug

* Move atomic fields into their own file

* Add a TODO

* Comments

* Remove metric.Labels(...)

* FTB

Co-authored-by: Liz Fong-Jones <lizf@honeycomb.io>
2020-03-27 14:06:48 -07:00

514 lines
13 KiB
Go

// Copyright The OpenTelemetry Authors
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
// This test is too large for the race detector. This SDK uses no locks
// that the race detector would help with, anyway.
// +build !race
package metric
import (
"context"
"fmt"
"math"
"math/rand"
"runtime"
"sort"
"strings"
"sync"
"sync/atomic"
"testing"
"time"
"go.opentelemetry.io/otel/api/core"
"go.opentelemetry.io/otel/api/key"
"go.opentelemetry.io/otel/api/metric"
api "go.opentelemetry.io/otel/api/metric"
export "go.opentelemetry.io/otel/sdk/export/metric"
"go.opentelemetry.io/otel/sdk/export/metric/aggregator"
"go.opentelemetry.io/otel/sdk/metric/aggregator/lastvalue"
"go.opentelemetry.io/otel/sdk/metric/aggregator/sum"
)
const (
concurrencyPerCPU = 100
reclaimPeriod = time.Millisecond * 100
testRun = 5 * time.Second
epsilon = 1e-10
)
var Must = api.Must
type (
testFixture struct {
// stop has to be aligned for 64-bit atomic operations.
stop int64
expected sync.Map
received sync.Map // Note: doesn't require synchronization
wg sync.WaitGroup
impl testImpl
T *testing.T
lock sync.Mutex
lused map[string]bool
dupCheck map[testKey]int
totalDups int64
}
testKey struct {
labels string
descriptor *metric.Descriptor
}
testImpl struct {
newInstrument func(meter api.Meter, name string) SyncImpler
getUpdateValue func() core.Number
operate func(interface{}, context.Context, core.Number, []core.KeyValue)
newStore func() interface{}
// storeCollect and storeExpect are the same for
// counters, different for lastValues, to ensure we are
// testing the timestamps correctly.
storeCollect func(store interface{}, value core.Number, ts time.Time)
storeExpect func(store interface{}, value core.Number)
readStore func(store interface{}) core.Number
equalValues func(a, b core.Number) bool
}
SyncImpler interface {
SyncImpl() metric.SyncImpl
}
// lastValueState supports merging lastValue values, for the case
// where a race condition causes duplicate records. We always
// take the later timestamp.
lastValueState struct {
// raw has to be aligned for 64-bit atomic operations.
raw core.Number
ts time.Time
}
)
func concurrency() int {
return concurrencyPerCPU * runtime.NumCPU()
}
func canonicalizeLabels(ls []core.KeyValue) string {
copy := append(ls[0:0:0], ls...)
sort.SliceStable(copy, func(i, j int) bool {
return copy[i].Key < copy[j].Key
})
var b strings.Builder
for _, kv := range copy {
b.WriteString(string(kv.Key))
b.WriteString("=")
b.WriteString(kv.Value.Emit())
b.WriteString("$")
}
return b.String()
}
func getPeriod() time.Duration {
dur := math.Max(
float64(reclaimPeriod)/10,
float64(reclaimPeriod)*(1+0.1*rand.NormFloat64()),
)
return time.Duration(dur)
}
func (f *testFixture) someLabels() []core.KeyValue {
n := 1 + rand.Intn(3)
l := make([]core.KeyValue, n)
for {
oused := map[string]bool{}
for i := 0; i < n; i++ {
var k string
for {
k = fmt.Sprint("k", rand.Intn(1000000000))
if !oused[k] {
oused[k] = true
break
}
}
l[i] = key.New(k).String(fmt.Sprint("v", rand.Intn(1000000000)))
}
lc := canonicalizeLabels(l)
f.lock.Lock()
avail := !f.lused[lc]
if avail {
f.lused[lc] = true
f.lock.Unlock()
return l
}
f.lock.Unlock()
}
}
func (f *testFixture) startWorker(impl *SDK, meter api.Meter, wg *sync.WaitGroup, i int) {
ctx := context.Background()
name := fmt.Sprint("test_", i)
instrument := f.impl.newInstrument(meter, name)
var descriptor *metric.Descriptor
if ii, ok := instrument.SyncImpl().(*syncInstrument); ok {
descriptor = &ii.descriptor
}
kvs := f.someLabels()
clabs := canonicalizeLabels(kvs)
dur := getPeriod()
key := testKey{
labels: clabs,
descriptor: descriptor,
}
for {
sleep := time.Duration(rand.ExpFloat64() * float64(dur))
time.Sleep(sleep)
value := f.impl.getUpdateValue()
f.impl.operate(instrument, ctx, value, kvs)
actual, _ := f.expected.LoadOrStore(key, f.impl.newStore())
f.impl.storeExpect(actual, value)
if atomic.LoadInt64(&f.stop) != 0 {
wg.Done()
return
}
}
}
func (f *testFixture) assertTest(numCollect int) {
var allErrs []func()
csize := 0
f.received.Range(func(key, gstore interface{}) bool {
csize++
gvalue := f.impl.readStore(gstore)
estore, loaded := f.expected.Load(key)
if !loaded {
allErrs = append(allErrs, func() {
f.T.Error("Could not locate expected key: ", key)
})
return true
}
evalue := f.impl.readStore(estore)
if !f.impl.equalValues(evalue, gvalue) {
allErrs = append(allErrs, func() {
f.T.Error("Expected value mismatch: ",
evalue, "!=", gvalue, " for ", key)
})
}
return true
})
rsize := 0
f.expected.Range(func(key, value interface{}) bool {
rsize++
if _, loaded := f.received.Load(key); !loaded {
allErrs = append(allErrs, func() {
f.T.Error("Did not receive expected key: ", key)
})
}
return true
})
if rsize != csize {
f.T.Error("Did not receive the correct set of metrics: Received != Expected", rsize, csize)
}
for _, anErr := range allErrs {
anErr()
}
// Note: It's useful to know the test triggers this condition,
// but we can't assert it. Infrequently no duplicates are
// found, and we can't really force a race to happen.
//
// fmt.Printf("Test duplicate records seen: %.1f%%\n",
// float64(100*f.totalDups/int64(numCollect*concurrency())))
}
func (f *testFixture) preCollect() {
// Collect calls Process in a single-threaded context. No need
// to lock this struct.
f.dupCheck = map[testKey]int{}
}
func (*testFixture) AggregatorFor(descriptor *metric.Descriptor) export.Aggregator {
name := descriptor.Name()
switch {
case strings.HasSuffix(name, "counter"):
return sum.New()
case strings.HasSuffix(name, "lastvalue"):
return lastvalue.New()
default:
panic("Not implemented for this test")
}
}
func (*testFixture) CheckpointSet() export.CheckpointSet {
return nil
}
func (*testFixture) FinishedCollection() {
}
func (f *testFixture) Process(_ context.Context, record export.Record) error {
labels := export.IteratorToSlice(record.Labels().Iter())
key := testKey{
labels: canonicalizeLabels(labels),
descriptor: record.Descriptor(),
}
if f.dupCheck[key] == 0 {
f.dupCheck[key]++
} else {
f.totalDups++
}
actual, _ := f.received.LoadOrStore(key, f.impl.newStore())
agg := record.Aggregator()
switch record.Descriptor().MetricKind() {
case metric.CounterKind:
sum, err := agg.(aggregator.Sum).Sum()
if err != nil {
f.T.Fatal("Sum error: ", err)
}
f.impl.storeCollect(actual, sum, time.Time{})
case metric.MeasureKind:
lv, ts, err := agg.(aggregator.LastValue).LastValue()
if err != nil && err != aggregator.ErrNoData {
f.T.Fatal("Last value error: ", err)
}
f.impl.storeCollect(actual, lv, ts)
default:
panic("Not used in this test")
}
return nil
}
func stressTest(t *testing.T, impl testImpl) {
ctx := context.Background()
t.Parallel()
fixture := &testFixture{
T: t,
impl: impl,
lused: map[string]bool{},
}
cc := concurrency()
sdk := New(fixture)
meter := metric.WrapMeterImpl(sdk, "stress_test")
fixture.wg.Add(cc + 1)
for i := 0; i < cc; i++ {
go fixture.startWorker(sdk, meter, &fixture.wg, i)
}
numCollect := 0
go func() {
for {
time.Sleep(reclaimPeriod)
fixture.preCollect()
sdk.Collect(ctx)
numCollect++
if atomic.LoadInt64(&fixture.stop) != 0 {
fixture.wg.Done()
return
}
}
}()
time.Sleep(testRun)
atomic.StoreInt64(&fixture.stop, 1)
fixture.wg.Wait()
fixture.preCollect()
sdk.Collect(ctx)
numCollect++
fixture.assertTest(numCollect)
}
func int64sEqual(a, b core.Number) bool {
return a.AsInt64() == b.AsInt64()
}
func float64sEqual(a, b core.Number) bool {
diff := math.Abs(a.AsFloat64() - b.AsFloat64())
return diff < math.Abs(a.AsFloat64())*epsilon
}
// Counters
func intCounterTestImpl() testImpl {
return testImpl{
newInstrument: func(meter api.Meter, name string) SyncImpler {
return Must(meter).NewInt64Counter(name + ".counter")
},
getUpdateValue: func() core.Number {
for {
x := int64(rand.Intn(100))
if x != 0 {
return core.NewInt64Number(x)
}
}
},
operate: func(inst interface{}, ctx context.Context, value core.Number, labels []core.KeyValue) {
counter := inst.(api.Int64Counter)
counter.Add(ctx, value.AsInt64(), labels...)
},
newStore: func() interface{} {
n := core.NewInt64Number(0)
return &n
},
storeCollect: func(store interface{}, value core.Number, _ time.Time) {
store.(*core.Number).AddInt64Atomic(value.AsInt64())
},
storeExpect: func(store interface{}, value core.Number) {
store.(*core.Number).AddInt64Atomic(value.AsInt64())
},
readStore: func(store interface{}) core.Number {
return store.(*core.Number).AsNumberAtomic()
},
equalValues: int64sEqual,
}
}
func TestStressInt64Counter(t *testing.T) {
stressTest(t, intCounterTestImpl())
}
func floatCounterTestImpl() testImpl {
return testImpl{
newInstrument: func(meter api.Meter, name string) SyncImpler {
return Must(meter).NewFloat64Counter(name + ".counter")
},
getUpdateValue: func() core.Number {
for {
x := rand.Float64()
if x != 0 {
return core.NewFloat64Number(x)
}
}
},
operate: func(inst interface{}, ctx context.Context, value core.Number, labels []core.KeyValue) {
counter := inst.(api.Float64Counter)
counter.Add(ctx, value.AsFloat64(), labels...)
},
newStore: func() interface{} {
n := core.NewFloat64Number(0.0)
return &n
},
storeCollect: func(store interface{}, value core.Number, _ time.Time) {
store.(*core.Number).AddFloat64Atomic(value.AsFloat64())
},
storeExpect: func(store interface{}, value core.Number) {
store.(*core.Number).AddFloat64Atomic(value.AsFloat64())
},
readStore: func(store interface{}) core.Number {
return store.(*core.Number).AsNumberAtomic()
},
equalValues: float64sEqual,
}
}
func TestStressFloat64Counter(t *testing.T) {
stressTest(t, floatCounterTestImpl())
}
// LastValue
func intLastValueTestImpl() testImpl {
return testImpl{
newInstrument: func(meter api.Meter, name string) SyncImpler {
return Must(meter).NewInt64Measure(name + ".lastvalue")
},
getUpdateValue: func() core.Number {
r1 := rand.Int63()
return core.NewInt64Number(rand.Int63() - r1)
},
operate: func(inst interface{}, ctx context.Context, value core.Number, labels []core.KeyValue) {
measure := inst.(api.Int64Measure)
measure.Record(ctx, value.AsInt64(), labels...)
},
newStore: func() interface{} {
return &lastValueState{
raw: core.NewInt64Number(0),
}
},
storeCollect: func(store interface{}, value core.Number, ts time.Time) {
gs := store.(*lastValueState)
if !ts.Before(gs.ts) {
gs.ts = ts
gs.raw.SetInt64Atomic(value.AsInt64())
}
},
storeExpect: func(store interface{}, value core.Number) {
gs := store.(*lastValueState)
gs.raw.SetInt64Atomic(value.AsInt64())
},
readStore: func(store interface{}) core.Number {
gs := store.(*lastValueState)
return gs.raw.AsNumberAtomic()
},
equalValues: int64sEqual,
}
}
func TestStressInt64LastValue(t *testing.T) {
stressTest(t, intLastValueTestImpl())
}
func floatLastValueTestImpl() testImpl {
return testImpl{
newInstrument: func(meter api.Meter, name string) SyncImpler {
return Must(meter).NewFloat64Measure(name + ".lastvalue")
},
getUpdateValue: func() core.Number {
return core.NewFloat64Number((-0.5 + rand.Float64()) * 100000)
},
operate: func(inst interface{}, ctx context.Context, value core.Number, labels []core.KeyValue) {
measure := inst.(api.Float64Measure)
measure.Record(ctx, value.AsFloat64(), labels...)
},
newStore: func() interface{} {
return &lastValueState{
raw: core.NewFloat64Number(0),
}
},
storeCollect: func(store interface{}, value core.Number, ts time.Time) {
gs := store.(*lastValueState)
if !ts.Before(gs.ts) {
gs.ts = ts
gs.raw.SetFloat64Atomic(value.AsFloat64())
}
},
storeExpect: func(store interface{}, value core.Number) {
gs := store.(*lastValueState)
gs.raw.SetFloat64Atomic(value.AsFloat64())
},
readStore: func(store interface{}) core.Number {
gs := store.(*lastValueState)
return gs.raw.AsNumberAtomic()
},
equalValues: float64sEqual,
}
}
func TestStressFloat64LastValue(t *testing.T) {
stressTest(t, floatLastValueTestImpl())
}