639 lines
12 KiB
Go
639 lines
12 KiB
Go
package models
|
|
|
|
import (
|
|
"testing"
|
|
"time"
|
|
|
|
"github.com/influxdata/telegraf"
|
|
"github.com/influxdata/telegraf/metric"
|
|
"github.com/influxdata/telegraf/testutil"
|
|
"github.com/stretchr/testify/require"
|
|
)
|
|
|
|
type MockMetric struct {
|
|
telegraf.Metric
|
|
AcceptF func()
|
|
RejectF func()
|
|
DropF func()
|
|
}
|
|
|
|
func (m *MockMetric) Accept() {
|
|
m.AcceptF()
|
|
}
|
|
|
|
func (m *MockMetric) Reject() {
|
|
m.RejectF()
|
|
}
|
|
|
|
func (m *MockMetric) Drop() {
|
|
m.DropF()
|
|
}
|
|
|
|
func Metric() telegraf.Metric {
|
|
return MetricTime(0)
|
|
}
|
|
|
|
func MetricTime(sec int64) telegraf.Metric {
|
|
m, err := metric.New(
|
|
"cpu",
|
|
map[string]string{},
|
|
map[string]interface{}{
|
|
"value": 42.0,
|
|
},
|
|
time.Unix(sec, 0),
|
|
)
|
|
if err != nil {
|
|
panic(err)
|
|
}
|
|
return m
|
|
}
|
|
|
|
func BenchmarkAddMetrics(b *testing.B) {
|
|
buf := NewBuffer("test", 10000)
|
|
m := Metric()
|
|
for n := 0; n < b.N; n++ {
|
|
buf.Add(m)
|
|
}
|
|
}
|
|
|
|
func setup(b *Buffer) *Buffer {
|
|
b.MetricsAdded.Set(0)
|
|
b.MetricsWritten.Set(0)
|
|
b.MetricsDropped.Set(0)
|
|
return b
|
|
}
|
|
|
|
func TestBuffer_LenEmpty(t *testing.T) {
|
|
b := setup(NewBuffer("test", 5))
|
|
|
|
require.Equal(t, 0, b.Len())
|
|
}
|
|
|
|
func TestBuffer_LenOne(t *testing.T) {
|
|
m := Metric()
|
|
b := setup(NewBuffer("test", 5))
|
|
b.Add(m)
|
|
|
|
require.Equal(t, 1, b.Len())
|
|
}
|
|
|
|
func TestBuffer_LenFull(t *testing.T) {
|
|
m := Metric()
|
|
b := setup(NewBuffer("test", 5))
|
|
b.Add(m, m, m, m, m)
|
|
|
|
require.Equal(t, 5, b.Len())
|
|
}
|
|
|
|
func TestBuffer_LenOverfill(t *testing.T) {
|
|
m := Metric()
|
|
b := setup(NewBuffer("test", 5))
|
|
setup(b)
|
|
b.Add(m, m, m, m, m, m)
|
|
|
|
require.Equal(t, 5, b.Len())
|
|
}
|
|
|
|
func TestBuffer_BatchLenZero(t *testing.T) {
|
|
b := setup(NewBuffer("test", 5))
|
|
batch := b.Batch(0)
|
|
|
|
require.Len(t, batch, 0)
|
|
}
|
|
|
|
func TestBuffer_BatchLenBufferEmpty(t *testing.T) {
|
|
b := setup(NewBuffer("test", 5))
|
|
batch := b.Batch(2)
|
|
|
|
require.Len(t, batch, 0)
|
|
}
|
|
|
|
func TestBuffer_BatchLenUnderfill(t *testing.T) {
|
|
m := Metric()
|
|
b := setup(NewBuffer("test", 5))
|
|
b.Add(m)
|
|
batch := b.Batch(2)
|
|
|
|
require.Len(t, batch, 1)
|
|
}
|
|
|
|
func TestBuffer_BatchLenFill(t *testing.T) {
|
|
m := Metric()
|
|
b := setup(NewBuffer("test", 5))
|
|
b.Add(m, m, m)
|
|
batch := b.Batch(2)
|
|
require.Len(t, batch, 2)
|
|
}
|
|
|
|
func TestBuffer_BatchLenExact(t *testing.T) {
|
|
m := Metric()
|
|
b := setup(NewBuffer("test", 5))
|
|
b.Add(m, m)
|
|
batch := b.Batch(2)
|
|
require.Len(t, batch, 2)
|
|
}
|
|
|
|
func TestBuffer_BatchLenLargerThanBuffer(t *testing.T) {
|
|
m := Metric()
|
|
b := setup(NewBuffer("test", 5))
|
|
b.Add(m, m, m, m, m)
|
|
batch := b.Batch(6)
|
|
require.Len(t, batch, 5)
|
|
}
|
|
|
|
func TestBuffer_BatchWrap(t *testing.T) {
|
|
m := Metric()
|
|
b := setup(NewBuffer("test", 5))
|
|
b.Add(m, m, m, m, m)
|
|
batch := b.Batch(2)
|
|
b.Accept(batch)
|
|
b.Add(m, m)
|
|
batch = b.Batch(5)
|
|
require.Len(t, batch, 5)
|
|
}
|
|
|
|
func TestBuffer_BatchLatest(t *testing.T) {
|
|
b := setup(NewBuffer("test", 4))
|
|
b.Add(MetricTime(1))
|
|
b.Add(MetricTime(2))
|
|
b.Add(MetricTime(3))
|
|
batch := b.Batch(2)
|
|
|
|
testutil.RequireMetricsEqual(t,
|
|
[]telegraf.Metric{
|
|
MetricTime(3),
|
|
MetricTime(2),
|
|
}, batch)
|
|
}
|
|
|
|
func TestBuffer_BatchLatestWrap(t *testing.T) {
|
|
b := setup(NewBuffer("test", 4))
|
|
b.Add(MetricTime(1))
|
|
b.Add(MetricTime(2))
|
|
b.Add(MetricTime(3))
|
|
b.Add(MetricTime(4))
|
|
b.Add(MetricTime(5))
|
|
batch := b.Batch(2)
|
|
|
|
testutil.RequireMetricsEqual(t,
|
|
[]telegraf.Metric{
|
|
MetricTime(5),
|
|
MetricTime(4),
|
|
}, batch)
|
|
}
|
|
|
|
func TestBuffer_MultipleBatch(t *testing.T) {
|
|
b := setup(NewBuffer("test", 10))
|
|
b.Add(MetricTime(1))
|
|
b.Add(MetricTime(2))
|
|
b.Add(MetricTime(3))
|
|
b.Add(MetricTime(4))
|
|
b.Add(MetricTime(5))
|
|
b.Add(MetricTime(6))
|
|
batch := b.Batch(5)
|
|
testutil.RequireMetricsEqual(t,
|
|
[]telegraf.Metric{
|
|
MetricTime(6),
|
|
MetricTime(5),
|
|
MetricTime(4),
|
|
MetricTime(3),
|
|
MetricTime(2),
|
|
}, batch)
|
|
b.Accept(batch)
|
|
batch = b.Batch(5)
|
|
testutil.RequireMetricsEqual(t,
|
|
[]telegraf.Metric{
|
|
MetricTime(1),
|
|
}, batch)
|
|
b.Accept(batch)
|
|
}
|
|
|
|
func TestBuffer_RejectWithRoom(t *testing.T) {
|
|
b := setup(NewBuffer("test", 5))
|
|
b.Add(MetricTime(1))
|
|
b.Add(MetricTime(2))
|
|
b.Add(MetricTime(3))
|
|
batch := b.Batch(2)
|
|
b.Add(MetricTime(4))
|
|
b.Add(MetricTime(5))
|
|
b.Reject(batch)
|
|
|
|
require.Equal(t, int64(0), b.MetricsDropped.Get())
|
|
|
|
batch = b.Batch(5)
|
|
testutil.RequireMetricsEqual(t,
|
|
[]telegraf.Metric{
|
|
MetricTime(5),
|
|
MetricTime(4),
|
|
MetricTime(3),
|
|
MetricTime(2),
|
|
MetricTime(1),
|
|
}, batch)
|
|
}
|
|
|
|
func TestBuffer_RejectNothingNewFull(t *testing.T) {
|
|
b := setup(NewBuffer("test", 5))
|
|
b.Add(MetricTime(1))
|
|
b.Add(MetricTime(2))
|
|
b.Add(MetricTime(3))
|
|
b.Add(MetricTime(4))
|
|
b.Add(MetricTime(5))
|
|
batch := b.Batch(2)
|
|
b.Reject(batch)
|
|
|
|
require.Equal(t, int64(0), b.MetricsDropped.Get())
|
|
|
|
batch = b.Batch(5)
|
|
testutil.RequireMetricsEqual(t,
|
|
[]telegraf.Metric{
|
|
MetricTime(5),
|
|
MetricTime(4),
|
|
MetricTime(3),
|
|
MetricTime(2),
|
|
MetricTime(1),
|
|
}, batch)
|
|
}
|
|
|
|
func TestBuffer_RejectNoRoom(t *testing.T) {
|
|
b := setup(NewBuffer("test", 5))
|
|
b.Add(MetricTime(1))
|
|
|
|
b.Add(MetricTime(2))
|
|
b.Add(MetricTime(3))
|
|
batch := b.Batch(2)
|
|
|
|
b.Add(MetricTime(4))
|
|
b.Add(MetricTime(5))
|
|
b.Add(MetricTime(6))
|
|
b.Add(MetricTime(7))
|
|
b.Add(MetricTime(8))
|
|
|
|
b.Reject(batch)
|
|
|
|
require.Equal(t, int64(3), b.MetricsDropped.Get())
|
|
|
|
batch = b.Batch(5)
|
|
testutil.RequireMetricsEqual(t,
|
|
[]telegraf.Metric{
|
|
MetricTime(8),
|
|
MetricTime(7),
|
|
MetricTime(6),
|
|
MetricTime(5),
|
|
MetricTime(4),
|
|
}, batch)
|
|
}
|
|
|
|
func TestBuffer_RejectRoomExact(t *testing.T) {
|
|
b := setup(NewBuffer("test", 5))
|
|
b.Add(MetricTime(1))
|
|
b.Add(MetricTime(2))
|
|
batch := b.Batch(2)
|
|
b.Add(MetricTime(3))
|
|
b.Add(MetricTime(4))
|
|
b.Add(MetricTime(5))
|
|
|
|
b.Reject(batch)
|
|
|
|
require.Equal(t, int64(0), b.MetricsDropped.Get())
|
|
|
|
batch = b.Batch(5)
|
|
testutil.RequireMetricsEqual(t,
|
|
[]telegraf.Metric{
|
|
MetricTime(5),
|
|
MetricTime(4),
|
|
MetricTime(3),
|
|
MetricTime(2),
|
|
MetricTime(1),
|
|
}, batch)
|
|
}
|
|
|
|
func TestBuffer_RejectRoomOverwriteOld(t *testing.T) {
|
|
b := setup(NewBuffer("test", 5))
|
|
b.Add(MetricTime(1))
|
|
b.Add(MetricTime(2))
|
|
b.Add(MetricTime(3))
|
|
batch := b.Batch(1)
|
|
b.Add(MetricTime(4))
|
|
b.Add(MetricTime(5))
|
|
b.Add(MetricTime(6))
|
|
|
|
b.Reject(batch)
|
|
|
|
require.Equal(t, int64(1), b.MetricsDropped.Get())
|
|
|
|
batch = b.Batch(5)
|
|
testutil.RequireMetricsEqual(t,
|
|
[]telegraf.Metric{
|
|
MetricTime(6),
|
|
MetricTime(5),
|
|
MetricTime(4),
|
|
MetricTime(3),
|
|
MetricTime(2),
|
|
}, batch)
|
|
}
|
|
|
|
func TestBuffer_RejectPartialRoom(t *testing.T) {
|
|
b := setup(NewBuffer("test", 5))
|
|
b.Add(MetricTime(1))
|
|
|
|
b.Add(MetricTime(2))
|
|
b.Add(MetricTime(3))
|
|
batch := b.Batch(2)
|
|
|
|
b.Add(MetricTime(4))
|
|
b.Add(MetricTime(5))
|
|
b.Add(MetricTime(6))
|
|
b.Add(MetricTime(7))
|
|
b.Reject(batch)
|
|
|
|
require.Equal(t, int64(2), b.MetricsDropped.Get())
|
|
|
|
batch = b.Batch(5)
|
|
testutil.RequireMetricsEqual(t,
|
|
[]telegraf.Metric{
|
|
MetricTime(7),
|
|
MetricTime(6),
|
|
MetricTime(5),
|
|
MetricTime(4),
|
|
MetricTime(3),
|
|
}, batch)
|
|
}
|
|
|
|
func TestBuffer_RejectWrapped(t *testing.T) {
|
|
b := setup(NewBuffer("test", 5))
|
|
b.Add(MetricTime(1))
|
|
b.Add(MetricTime(2))
|
|
b.Add(MetricTime(3))
|
|
batch := b.Batch(2)
|
|
b.Add(MetricTime(4))
|
|
b.Add(MetricTime(5))
|
|
|
|
// buffer: 1, 4, 5; batch: 2, 3
|
|
require.Equal(t, int64(0), b.MetricsDropped.Get())
|
|
|
|
b.Add(MetricTime(6))
|
|
b.Add(MetricTime(7))
|
|
b.Add(MetricTime(8))
|
|
b.Add(MetricTime(9))
|
|
b.Add(MetricTime(10))
|
|
|
|
// buffer: 8, 9, 10, 6, 7; batch: 2, 3
|
|
require.Equal(t, int64(3), b.MetricsDropped.Get())
|
|
|
|
b.Add(MetricTime(11))
|
|
b.Add(MetricTime(12))
|
|
b.Add(MetricTime(13))
|
|
b.Add(MetricTime(14))
|
|
b.Add(MetricTime(15))
|
|
// buffer: 13, 14, 15, 11, 12; batch: 2, 3
|
|
require.Equal(t, int64(8), b.MetricsDropped.Get())
|
|
b.Reject(batch)
|
|
|
|
require.Equal(t, int64(10), b.MetricsDropped.Get())
|
|
|
|
batch = b.Batch(5)
|
|
testutil.RequireMetricsEqual(t,
|
|
[]telegraf.Metric{
|
|
MetricTime(15),
|
|
MetricTime(14),
|
|
MetricTime(13),
|
|
MetricTime(12),
|
|
MetricTime(11),
|
|
}, batch)
|
|
}
|
|
|
|
func TestBuffer_AddDropsOverwrittenMetrics(t *testing.T) {
|
|
m := Metric()
|
|
b := setup(NewBuffer("test", 5))
|
|
|
|
b.Add(m, m, m, m, m)
|
|
b.Add(m, m, m, m, m)
|
|
|
|
require.Equal(t, int64(5), b.MetricsDropped.Get())
|
|
require.Equal(t, int64(0), b.MetricsWritten.Get())
|
|
}
|
|
|
|
func TestBuffer_AcceptRemovesBatch(t *testing.T) {
|
|
m := Metric()
|
|
b := setup(NewBuffer("test", 5))
|
|
b.Add(m, m, m)
|
|
batch := b.Batch(2)
|
|
b.Accept(batch)
|
|
require.Equal(t, 1, b.Len())
|
|
}
|
|
|
|
func TestBuffer_RejectLeavesBatch(t *testing.T) {
|
|
m := Metric()
|
|
b := setup(NewBuffer("test", 5))
|
|
b.Add(m, m, m)
|
|
batch := b.Batch(2)
|
|
b.Reject(batch)
|
|
require.Equal(t, 3, b.Len())
|
|
}
|
|
|
|
func TestBuffer_AcceptWritesOverwrittenBatch(t *testing.T) {
|
|
m := Metric()
|
|
b := setup(NewBuffer("test", 5))
|
|
|
|
b.Add(m, m, m, m, m)
|
|
batch := b.Batch(5)
|
|
b.Add(m, m, m, m, m)
|
|
b.Accept(batch)
|
|
|
|
require.Equal(t, int64(0), b.MetricsDropped.Get())
|
|
require.Equal(t, int64(5), b.MetricsWritten.Get())
|
|
}
|
|
|
|
func TestBuffer_BatchRejectDropsOverwrittenBatch(t *testing.T) {
|
|
m := Metric()
|
|
b := setup(NewBuffer("test", 5))
|
|
|
|
b.Add(m, m, m, m, m)
|
|
batch := b.Batch(5)
|
|
b.Add(m, m, m, m, m)
|
|
b.Reject(batch)
|
|
|
|
require.Equal(t, int64(5), b.MetricsDropped.Get())
|
|
require.Equal(t, int64(0), b.MetricsWritten.Get())
|
|
}
|
|
|
|
func TestBuffer_MetricsOverwriteBatchAccept(t *testing.T) {
|
|
m := Metric()
|
|
b := setup(NewBuffer("test", 5))
|
|
|
|
b.Add(m, m, m, m, m)
|
|
batch := b.Batch(3)
|
|
b.Add(m, m, m)
|
|
b.Accept(batch)
|
|
require.Equal(t, int64(0), b.MetricsDropped.Get(), "dropped")
|
|
require.Equal(t, int64(3), b.MetricsWritten.Get(), "written")
|
|
}
|
|
|
|
func TestBuffer_MetricsOverwriteBatchReject(t *testing.T) {
|
|
m := Metric()
|
|
b := setup(NewBuffer("test", 5))
|
|
|
|
b.Add(m, m, m, m, m)
|
|
batch := b.Batch(3)
|
|
b.Add(m, m, m)
|
|
b.Reject(batch)
|
|
require.Equal(t, int64(3), b.MetricsDropped.Get())
|
|
require.Equal(t, int64(0), b.MetricsWritten.Get())
|
|
}
|
|
|
|
func TestBuffer_MetricsBatchAcceptRemoved(t *testing.T) {
|
|
m := Metric()
|
|
b := setup(NewBuffer("test", 5))
|
|
|
|
b.Add(m, m, m, m, m)
|
|
batch := b.Batch(3)
|
|
b.Add(m, m, m, m, m)
|
|
b.Accept(batch)
|
|
require.Equal(t, int64(2), b.MetricsDropped.Get())
|
|
require.Equal(t, int64(3), b.MetricsWritten.Get())
|
|
}
|
|
|
|
func TestBuffer_WrapWithBatch(t *testing.T) {
|
|
m := Metric()
|
|
b := setup(NewBuffer("test", 5))
|
|
|
|
b.Add(m, m, m)
|
|
b.Batch(3)
|
|
b.Add(m, m, m, m, m, m)
|
|
|
|
require.Equal(t, int64(1), b.MetricsDropped.Get())
|
|
}
|
|
|
|
func TestBuffer_BatchNotRemoved(t *testing.T) {
|
|
m := Metric()
|
|
b := setup(NewBuffer("test", 5))
|
|
b.Add(m, m, m, m, m)
|
|
b.Batch(2)
|
|
require.Equal(t, 3, b.Len())
|
|
}
|
|
|
|
func TestBuffer_BatchRejectAcceptNoop(t *testing.T) {
|
|
m := Metric()
|
|
b := setup(NewBuffer("test", 5))
|
|
b.Add(m, m, m, m, m)
|
|
batch := b.Batch(2)
|
|
b.Reject(batch)
|
|
b.Accept(batch)
|
|
require.Equal(t, 5, b.Len())
|
|
}
|
|
|
|
func TestBuffer_AcceptCallsMetricAccept(t *testing.T) {
|
|
var accept int
|
|
mm := &MockMetric{
|
|
Metric: Metric(),
|
|
AcceptF: func() {
|
|
accept++
|
|
},
|
|
}
|
|
b := setup(NewBuffer("test", 5))
|
|
b.Add(mm, mm, mm)
|
|
batch := b.Batch(2)
|
|
b.Accept(batch)
|
|
require.Equal(t, 2, accept)
|
|
}
|
|
|
|
func TestBuffer_AddCallsMetricRejectWhenNoBatch(t *testing.T) {
|
|
var reject int
|
|
mm := &MockMetric{
|
|
Metric: Metric(),
|
|
RejectF: func() {
|
|
reject++
|
|
},
|
|
}
|
|
b := setup(NewBuffer("test", 5))
|
|
setup(b)
|
|
b.Add(mm, mm, mm, mm, mm)
|
|
b.Add(mm, mm)
|
|
require.Equal(t, 2, reject)
|
|
}
|
|
|
|
func TestBuffer_AddCallsMetricRejectWhenNotInBatch(t *testing.T) {
|
|
var reject int
|
|
mm := &MockMetric{
|
|
Metric: Metric(),
|
|
RejectF: func() {
|
|
reject++
|
|
},
|
|
}
|
|
b := setup(NewBuffer("test", 5))
|
|
setup(b)
|
|
b.Add(mm, mm, mm, mm, mm)
|
|
batch := b.Batch(2)
|
|
b.Add(mm, mm, mm, mm)
|
|
require.Equal(t, 2, reject)
|
|
b.Reject(batch)
|
|
require.Equal(t, 4, reject)
|
|
}
|
|
|
|
func TestBuffer_RejectCallsMetricRejectWithOverwritten(t *testing.T) {
|
|
var reject int
|
|
mm := &MockMetric{
|
|
Metric: Metric(),
|
|
RejectF: func() {
|
|
reject++
|
|
},
|
|
}
|
|
b := setup(NewBuffer("test", 5))
|
|
b.Add(mm, mm, mm, mm, mm)
|
|
batch := b.Batch(5)
|
|
b.Add(mm, mm)
|
|
require.Equal(t, 0, reject)
|
|
b.Reject(batch)
|
|
require.Equal(t, 2, reject)
|
|
}
|
|
|
|
func TestBuffer_AddOverwriteAndReject(t *testing.T) {
|
|
var reject int
|
|
mm := &MockMetric{
|
|
Metric: Metric(),
|
|
RejectF: func() {
|
|
reject++
|
|
},
|
|
}
|
|
b := setup(NewBuffer("test", 5))
|
|
b.Add(mm, mm, mm, mm, mm)
|
|
batch := b.Batch(5)
|
|
b.Add(mm, mm, mm, mm, mm)
|
|
b.Add(mm, mm, mm, mm, mm)
|
|
b.Add(mm, mm, mm, mm, mm)
|
|
b.Add(mm, mm, mm, mm, mm)
|
|
require.Equal(t, 15, reject)
|
|
b.Reject(batch)
|
|
require.Equal(t, 20, reject)
|
|
}
|
|
|
|
func TestBuffer_AddOverwriteAndRejectOffset(t *testing.T) {
|
|
var reject int
|
|
var accept int
|
|
mm := &MockMetric{
|
|
Metric: Metric(),
|
|
RejectF: func() {
|
|
reject++
|
|
},
|
|
AcceptF: func() {
|
|
accept++
|
|
},
|
|
}
|
|
b := setup(NewBuffer("test", 5))
|
|
b.Add(mm, mm, mm)
|
|
b.Add(mm, mm, mm, mm)
|
|
require.Equal(t, 2, reject)
|
|
batch := b.Batch(5)
|
|
b.Add(mm, mm, mm, mm)
|
|
require.Equal(t, 2, reject)
|
|
b.Add(mm, mm, mm, mm)
|
|
require.Equal(t, 5, reject)
|
|
b.Add(mm, mm, mm, mm)
|
|
require.Equal(t, 9, reject)
|
|
b.Add(mm, mm, mm, mm)
|
|
require.Equal(t, 13, reject)
|
|
b.Accept(batch)
|
|
require.Equal(t, 13, reject)
|
|
require.Equal(t, 5, accept)
|
|
}
|