mirror of
https://github.com/prometheus/prometheus.git
synced 2024-12-30 07:59:40 -08:00
c328d5fc88
is reported for these benchmarks to more accurately reflect what's actually allocated Signed-off-by: Callum Styan <callumstyan@gmail.com>
2292 lines
71 KiB
Go
2292 lines
71 KiB
Go
// Copyright 2013 The Prometheus 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.
|
|
|
|
package remote
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"fmt"
|
|
"math"
|
|
"math/rand"
|
|
"os"
|
|
"runtime/pprof"
|
|
"sort"
|
|
"strconv"
|
|
"strings"
|
|
"sync"
|
|
"testing"
|
|
"time"
|
|
|
|
"github.com/go-kit/log"
|
|
"github.com/gogo/protobuf/proto"
|
|
"github.com/golang/snappy"
|
|
"github.com/google/go-cmp/cmp"
|
|
"github.com/prometheus/client_golang/prometheus"
|
|
client_testutil "github.com/prometheus/client_golang/prometheus/testutil"
|
|
"github.com/prometheus/common/model"
|
|
"github.com/stretchr/testify/require"
|
|
"go.uber.org/atomic"
|
|
|
|
"github.com/prometheus/prometheus/config"
|
|
"github.com/prometheus/prometheus/model/histogram"
|
|
"github.com/prometheus/prometheus/model/labels"
|
|
"github.com/prometheus/prometheus/model/relabel"
|
|
"github.com/prometheus/prometheus/model/timestamp"
|
|
"github.com/prometheus/prometheus/prompb"
|
|
writev2 "github.com/prometheus/prometheus/prompb/io/prometheus/write/v2"
|
|
"github.com/prometheus/prometheus/scrape"
|
|
"github.com/prometheus/prometheus/tsdb/chunks"
|
|
"github.com/prometheus/prometheus/tsdb/record"
|
|
"github.com/prometheus/prometheus/util/runutil"
|
|
"github.com/prometheus/prometheus/util/testutil"
|
|
)
|
|
|
|
const defaultFlushDeadline = 1 * time.Minute
|
|
|
|
func newHighestTimestampMetric() *maxTimestamp {
|
|
return &maxTimestamp{
|
|
Gauge: prometheus.NewGauge(prometheus.GaugeOpts{
|
|
Namespace: namespace,
|
|
Subsystem: subsystem,
|
|
Name: "highest_timestamp_in_seconds",
|
|
Help: "Highest timestamp that has come into the remote storage via the Appender interface, in seconds since epoch. Initialized to 0 when no data has been received yet",
|
|
}),
|
|
}
|
|
}
|
|
|
|
func TestBasicContentNegotiation(t *testing.T) {
|
|
queueConfig := config.DefaultQueueConfig
|
|
queueConfig.BatchSendDeadline = model.Duration(100 * time.Millisecond)
|
|
queueConfig.MaxShards = 1
|
|
|
|
// We need to set URL's so that metric creation doesn't panic.
|
|
writeConfig := baseRemoteWriteConfig("http://test-storage.com")
|
|
writeConfig.QueueConfig = queueConfig
|
|
|
|
conf := &config.Config{
|
|
GlobalConfig: config.DefaultGlobalConfig,
|
|
RemoteWriteConfigs: []*config.RemoteWriteConfig{
|
|
writeConfig,
|
|
},
|
|
}
|
|
|
|
for _, tc := range []struct {
|
|
name string
|
|
senderProtoMsg config.RemoteWriteProtoMsg
|
|
receiverProtoMsg config.RemoteWriteProtoMsg
|
|
injectErrs []error
|
|
expectFail bool
|
|
}{
|
|
{
|
|
name: "v2 happy path",
|
|
senderProtoMsg: config.RemoteWriteProtoMsgV2, receiverProtoMsg: config.RemoteWriteProtoMsgV2,
|
|
injectErrs: []error{nil},
|
|
},
|
|
{
|
|
name: "v1 happy path",
|
|
senderProtoMsg: config.RemoteWriteProtoMsgV1, receiverProtoMsg: config.RemoteWriteProtoMsgV1,
|
|
injectErrs: []error{nil},
|
|
},
|
|
// Test a case where the v1 request has a temporary delay but goes through on retry.
|
|
{
|
|
name: "v1 happy path with one 5xx retry",
|
|
senderProtoMsg: config.RemoteWriteProtoMsgV1, receiverProtoMsg: config.RemoteWriteProtoMsgV1,
|
|
injectErrs: []error{RecoverableError{errors.New("pretend 500"), 1}, nil},
|
|
},
|
|
// Repeat the above test but with v2. The request has a temporary delay but goes through on retry.
|
|
{
|
|
name: "v2 happy path with one 5xx retry",
|
|
senderProtoMsg: config.RemoteWriteProtoMsgV2, receiverProtoMsg: config.RemoteWriteProtoMsgV2,
|
|
injectErrs: []error{RecoverableError{errors.New("pretend 500"), 1}, nil},
|
|
},
|
|
// A few error cases of v2 talking to v1.
|
|
{
|
|
name: "v2 talks to v1 that gives 400 or 415",
|
|
senderProtoMsg: config.RemoteWriteProtoMsgV2, receiverProtoMsg: config.RemoteWriteProtoMsgV1,
|
|
injectErrs: []error{errors.New("pretend unrecoverable err")},
|
|
expectFail: true,
|
|
},
|
|
{
|
|
name: "v2 talks to (broken) v1 that tries to unmarshal v2 payload with v1 proto",
|
|
senderProtoMsg: config.RemoteWriteProtoMsgV2, receiverProtoMsg: config.RemoteWriteProtoMsgV1,
|
|
injectErrs: []error{nil},
|
|
expectFail: true, // We detect this thanks to https://github.com/prometheus/prometheus/issues/14359
|
|
},
|
|
// Opposite, v1 talking to v2 only server.
|
|
{
|
|
name: "v1 talks to v2 that gives 400 or 415",
|
|
senderProtoMsg: config.RemoteWriteProtoMsgV1, receiverProtoMsg: config.RemoteWriteProtoMsgV2,
|
|
injectErrs: []error{errors.New("pretend unrecoverable err")},
|
|
expectFail: true,
|
|
},
|
|
} {
|
|
t.Run(tc.name, func(t *testing.T) {
|
|
dir := t.TempDir()
|
|
s := NewStorage(nil, nil, nil, dir, defaultFlushDeadline, nil, true)
|
|
defer s.Close()
|
|
|
|
var (
|
|
series []record.RefSeries
|
|
metadata []record.RefMetadata
|
|
samples []record.RefSample
|
|
)
|
|
|
|
// Generates same series in both cases.
|
|
samples, series = createTimeseries(1, 1)
|
|
metadata = createSeriesMetadata(series)
|
|
|
|
// Apply new config.
|
|
queueConfig.Capacity = len(samples)
|
|
queueConfig.MaxSamplesPerSend = len(samples)
|
|
// For now we only ever have a single rw config in this test.
|
|
conf.RemoteWriteConfigs[0].ProtobufMessage = tc.senderProtoMsg
|
|
require.NoError(t, s.ApplyConfig(conf))
|
|
hash, err := toHash(writeConfig)
|
|
require.NoError(t, err)
|
|
qm := s.rws.queues[hash]
|
|
|
|
c := NewTestWriteClient(tc.receiverProtoMsg)
|
|
c.injectErrors(tc.injectErrs)
|
|
qm.SetClient(c)
|
|
|
|
qm.StoreSeries(series, 0)
|
|
qm.StoreMetadata(metadata)
|
|
|
|
// Do we expect some data back?
|
|
if !tc.expectFail {
|
|
c.expectSamples(samples, series)
|
|
} else {
|
|
c.expectSamples(nil, nil)
|
|
}
|
|
|
|
// Schedule send.
|
|
qm.Append(samples)
|
|
|
|
if !tc.expectFail {
|
|
// No error expected, so wait for data.
|
|
c.waitForExpectedData(t, 5*time.Second)
|
|
require.Equal(t, 0.0, client_testutil.ToFloat64(qm.metrics.failedSamplesTotal))
|
|
} else {
|
|
// Wait for failure to be recorded in metrics.
|
|
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
|
|
defer cancel()
|
|
require.NoError(t, runutil.Retry(500*time.Millisecond, ctx.Done(), func() error {
|
|
if client_testutil.ToFloat64(qm.metrics.failedSamplesTotal) != 1.0 {
|
|
return fmt.Errorf("expected one sample failed in qm metrics; got %v", client_testutil.ToFloat64(qm.metrics.failedSamplesTotal))
|
|
}
|
|
return nil
|
|
}))
|
|
}
|
|
|
|
// samplesTotal means attempts.
|
|
require.Equal(t, float64(len(tc.injectErrs)), client_testutil.ToFloat64(qm.metrics.samplesTotal))
|
|
require.Equal(t, float64(len(tc.injectErrs)-1), client_testutil.ToFloat64(qm.metrics.retriedSamplesTotal))
|
|
})
|
|
}
|
|
}
|
|
|
|
func TestSampleDelivery(t *testing.T) {
|
|
// Let's create an even number of send batches, so we don't run into the
|
|
// batch timeout case.
|
|
n := 3
|
|
|
|
queueConfig := config.DefaultQueueConfig
|
|
queueConfig.BatchSendDeadline = model.Duration(100 * time.Millisecond)
|
|
queueConfig.MaxShards = 1
|
|
|
|
// We need to set URL's so that metric creation doesn't panic.
|
|
writeConfig := baseRemoteWriteConfig("http://test-storage.com")
|
|
writeConfig.QueueConfig = queueConfig
|
|
writeConfig.SendExemplars = true
|
|
writeConfig.SendNativeHistograms = true
|
|
|
|
conf := &config.Config{
|
|
GlobalConfig: config.DefaultGlobalConfig,
|
|
RemoteWriteConfigs: []*config.RemoteWriteConfig{
|
|
writeConfig,
|
|
},
|
|
}
|
|
for _, tc := range []struct {
|
|
protoMsg config.RemoteWriteProtoMsg
|
|
|
|
name string
|
|
samples bool
|
|
exemplars bool
|
|
histograms bool
|
|
floatHistograms bool
|
|
}{
|
|
{protoMsg: config.RemoteWriteProtoMsgV1, samples: true, exemplars: false, histograms: false, floatHistograms: false, name: "samples only"},
|
|
{protoMsg: config.RemoteWriteProtoMsgV1, samples: true, exemplars: true, histograms: true, floatHistograms: true, name: "samples, exemplars, and histograms"},
|
|
{protoMsg: config.RemoteWriteProtoMsgV1, samples: false, exemplars: true, histograms: false, floatHistograms: false, name: "exemplars only"},
|
|
{protoMsg: config.RemoteWriteProtoMsgV1, samples: false, exemplars: false, histograms: true, floatHistograms: false, name: "histograms only"},
|
|
{protoMsg: config.RemoteWriteProtoMsgV1, samples: false, exemplars: false, histograms: false, floatHistograms: true, name: "float histograms only"},
|
|
|
|
// TODO(alexg): update some portion of this test to check for the 2.0 metadata
|
|
{protoMsg: config.RemoteWriteProtoMsgV2, samples: true, exemplars: false, histograms: false, floatHistograms: false, name: "samples only"},
|
|
{protoMsg: config.RemoteWriteProtoMsgV2, samples: true, exemplars: true, histograms: true, floatHistograms: true, name: "samples, exemplars, and histograms"},
|
|
{protoMsg: config.RemoteWriteProtoMsgV2, samples: false, exemplars: true, histograms: false, floatHistograms: false, name: "exemplars only"},
|
|
{protoMsg: config.RemoteWriteProtoMsgV2, samples: false, exemplars: false, histograms: true, floatHistograms: false, name: "histograms only"},
|
|
{protoMsg: config.RemoteWriteProtoMsgV2, samples: false, exemplars: false, histograms: false, floatHistograms: true, name: "float histograms only"},
|
|
} {
|
|
t.Run(fmt.Sprintf("%s-%s", tc.protoMsg, tc.name), func(t *testing.T) {
|
|
dir := t.TempDir()
|
|
s := NewStorage(nil, nil, nil, dir, defaultFlushDeadline, nil, true)
|
|
defer s.Close()
|
|
|
|
var (
|
|
series []record.RefSeries
|
|
metadata []record.RefMetadata
|
|
samples []record.RefSample
|
|
exemplars []record.RefExemplar
|
|
histograms []record.RefHistogramSample
|
|
floatHistograms []record.RefFloatHistogramSample
|
|
)
|
|
|
|
// Generates same series in both cases.
|
|
if tc.samples {
|
|
samples, series = createTimeseries(n, n)
|
|
}
|
|
if tc.exemplars {
|
|
exemplars, series = createExemplars(n, n)
|
|
}
|
|
if tc.histograms {
|
|
histograms, _, series = createHistograms(n, n, false)
|
|
}
|
|
if tc.floatHistograms {
|
|
_, floatHistograms, series = createHistograms(n, n, true)
|
|
}
|
|
metadata = createSeriesMetadata(series)
|
|
|
|
// Apply new config.
|
|
queueConfig.Capacity = len(samples)
|
|
queueConfig.MaxSamplesPerSend = len(samples) / 2
|
|
// For now we only ever have a single rw config in this test.
|
|
conf.RemoteWriteConfigs[0].ProtobufMessage = tc.protoMsg
|
|
require.NoError(t, s.ApplyConfig(conf))
|
|
hash, err := toHash(writeConfig)
|
|
require.NoError(t, err)
|
|
qm := s.rws.queues[hash]
|
|
|
|
c := NewTestWriteClient(tc.protoMsg)
|
|
qm.SetClient(c)
|
|
|
|
qm.StoreSeries(series, 0)
|
|
qm.StoreMetadata(metadata)
|
|
|
|
// Send first half of data.
|
|
c.expectSamples(samples[:len(samples)/2], series)
|
|
c.expectExemplars(exemplars[:len(exemplars)/2], series)
|
|
c.expectHistograms(histograms[:len(histograms)/2], series)
|
|
c.expectFloatHistograms(floatHistograms[:len(floatHistograms)/2], series)
|
|
qm.Append(samples[:len(samples)/2])
|
|
qm.AppendExemplars(exemplars[:len(exemplars)/2])
|
|
qm.AppendHistograms(histograms[:len(histograms)/2])
|
|
qm.AppendFloatHistograms(floatHistograms[:len(floatHistograms)/2])
|
|
c.waitForExpectedData(t, 30*time.Second)
|
|
|
|
// Send second half of data.
|
|
c.expectSamples(samples[len(samples)/2:], series)
|
|
c.expectExemplars(exemplars[len(exemplars)/2:], series)
|
|
c.expectHistograms(histograms[len(histograms)/2:], series)
|
|
c.expectFloatHistograms(floatHistograms[len(floatHistograms)/2:], series)
|
|
qm.Append(samples[len(samples)/2:])
|
|
qm.AppendExemplars(exemplars[len(exemplars)/2:])
|
|
qm.AppendHistograms(histograms[len(histograms)/2:])
|
|
qm.AppendFloatHistograms(floatHistograms[len(floatHistograms)/2:])
|
|
c.waitForExpectedData(t, 30*time.Second)
|
|
})
|
|
}
|
|
}
|
|
|
|
func newTestClientAndQueueManager(t testing.TB, flushDeadline time.Duration, protoMsg config.RemoteWriteProtoMsg) (*TestWriteClient, *QueueManager) {
|
|
c := NewTestWriteClient(protoMsg)
|
|
cfg := config.DefaultQueueConfig
|
|
mcfg := config.DefaultMetadataConfig
|
|
return c, newTestQueueManager(t, cfg, mcfg, flushDeadline, c, protoMsg)
|
|
}
|
|
|
|
func newTestQueueManager(t testing.TB, cfg config.QueueConfig, mcfg config.MetadataConfig, deadline time.Duration, c WriteClient, protoMsg config.RemoteWriteProtoMsg) *QueueManager {
|
|
dir := t.TempDir()
|
|
metrics := newQueueManagerMetrics(nil, "", "")
|
|
m := NewQueueManager(metrics, nil, nil, nil, dir, newEWMARate(ewmaWeight, shardUpdateDuration), cfg, mcfg, labels.EmptyLabels(), nil, c, deadline, newPool(), newHighestTimestampMetric(), nil, false, false, protoMsg)
|
|
|
|
return m
|
|
}
|
|
|
|
func testDefaultQueueConfig() config.QueueConfig {
|
|
cfg := config.DefaultQueueConfig
|
|
// For faster unit tests we don't wait default 5 seconds.
|
|
cfg.BatchSendDeadline = model.Duration(100 * time.Millisecond)
|
|
return cfg
|
|
}
|
|
|
|
func TestMetadataDelivery(t *testing.T) {
|
|
c, m := newTestClientAndQueueManager(t, defaultFlushDeadline, config.RemoteWriteProtoMsgV1)
|
|
m.Start()
|
|
defer m.Stop()
|
|
|
|
metadata := []scrape.MetricMetadata{}
|
|
numMetadata := 1532
|
|
for i := 0; i < numMetadata; i++ {
|
|
metadata = append(metadata, scrape.MetricMetadata{
|
|
Metric: "prometheus_remote_storage_sent_metadata_bytes_total_" + strconv.Itoa(i),
|
|
Type: model.MetricTypeCounter,
|
|
Help: "a nice help text",
|
|
Unit: "",
|
|
})
|
|
}
|
|
|
|
m.AppendWatcherMetadata(context.Background(), metadata)
|
|
|
|
require.Equal(t, 0.0, client_testutil.ToFloat64(m.metrics.failedMetadataTotal))
|
|
require.Len(t, c.receivedMetadata, numMetadata)
|
|
// One more write than the rounded quotient should be performed in order to get samples that didn't
|
|
// fit into MaxSamplesPerSend.
|
|
require.Equal(t, numMetadata/config.DefaultMetadataConfig.MaxSamplesPerSend+1, c.writesReceived)
|
|
// Make sure the last samples were sent.
|
|
require.Equal(t, c.receivedMetadata[metadata[len(metadata)-1].Metric][0].MetricFamilyName, metadata[len(metadata)-1].Metric)
|
|
}
|
|
|
|
func TestWALMetadataDelivery(t *testing.T) {
|
|
dir := t.TempDir()
|
|
s := NewStorage(nil, nil, nil, dir, defaultFlushDeadline, nil, true)
|
|
defer s.Close()
|
|
|
|
cfg := config.DefaultQueueConfig
|
|
cfg.BatchSendDeadline = model.Duration(100 * time.Millisecond)
|
|
cfg.MaxShards = 1
|
|
|
|
writeConfig := baseRemoteWriteConfig("http://test-storage.com")
|
|
writeConfig.QueueConfig = cfg
|
|
writeConfig.ProtobufMessage = config.RemoteWriteProtoMsgV2
|
|
|
|
conf := &config.Config{
|
|
GlobalConfig: config.DefaultGlobalConfig,
|
|
RemoteWriteConfigs: []*config.RemoteWriteConfig{
|
|
writeConfig,
|
|
},
|
|
}
|
|
|
|
num := 3
|
|
_, series := createTimeseries(0, num)
|
|
metadata := createSeriesMetadata(series)
|
|
|
|
require.NoError(t, s.ApplyConfig(conf))
|
|
hash, err := toHash(writeConfig)
|
|
require.NoError(t, err)
|
|
qm := s.rws.queues[hash]
|
|
|
|
c := NewTestWriteClient(config.RemoteWriteProtoMsgV1)
|
|
qm.SetClient(c)
|
|
|
|
qm.StoreSeries(series, 0)
|
|
qm.StoreMetadata(metadata)
|
|
|
|
require.Len(t, qm.seriesLabels, num)
|
|
require.Len(t, qm.seriesMetadata, num)
|
|
|
|
c.waitForExpectedData(t, 30*time.Second)
|
|
}
|
|
|
|
func TestSampleDeliveryTimeout(t *testing.T) {
|
|
for _, protoMsg := range []config.RemoteWriteProtoMsg{config.RemoteWriteProtoMsgV1, config.RemoteWriteProtoMsgV2} {
|
|
t.Run(fmt.Sprint(protoMsg), func(t *testing.T) {
|
|
// Let's send one less sample than batch size, and wait the timeout duration
|
|
n := 9
|
|
samples, series := createTimeseries(n, n)
|
|
cfg := testDefaultQueueConfig()
|
|
mcfg := config.DefaultMetadataConfig
|
|
cfg.MaxShards = 1
|
|
|
|
c := NewTestWriteClient(protoMsg)
|
|
m := newTestQueueManager(t, cfg, mcfg, defaultFlushDeadline, c, protoMsg)
|
|
m.StoreSeries(series, 0)
|
|
m.Start()
|
|
defer m.Stop()
|
|
|
|
// Send the samples twice, waiting for the samples in the meantime.
|
|
c.expectSamples(samples, series)
|
|
m.Append(samples)
|
|
c.waitForExpectedData(t, 30*time.Second)
|
|
|
|
c.expectSamples(samples, series)
|
|
m.Append(samples)
|
|
c.waitForExpectedData(t, 30*time.Second)
|
|
})
|
|
}
|
|
}
|
|
|
|
func TestSampleDeliveryOrder(t *testing.T) {
|
|
for _, protoMsg := range []config.RemoteWriteProtoMsg{config.RemoteWriteProtoMsgV1, config.RemoteWriteProtoMsgV2} {
|
|
t.Run(fmt.Sprint(protoMsg), func(t *testing.T) {
|
|
ts := 10
|
|
n := config.DefaultQueueConfig.MaxSamplesPerSend * ts
|
|
samples := make([]record.RefSample, 0, n)
|
|
series := make([]record.RefSeries, 0, n)
|
|
for i := 0; i < n; i++ {
|
|
name := fmt.Sprintf("test_metric_%d", i%ts)
|
|
samples = append(samples, record.RefSample{
|
|
Ref: chunks.HeadSeriesRef(i),
|
|
T: int64(i),
|
|
V: float64(i),
|
|
})
|
|
series = append(series, record.RefSeries{
|
|
Ref: chunks.HeadSeriesRef(i),
|
|
Labels: labels.FromStrings("__name__", name),
|
|
})
|
|
}
|
|
|
|
c, m := newTestClientAndQueueManager(t, defaultFlushDeadline, protoMsg)
|
|
c.expectSamples(samples, series)
|
|
m.StoreSeries(series, 0)
|
|
|
|
m.Start()
|
|
defer m.Stop()
|
|
// These should be received by the client.
|
|
m.Append(samples)
|
|
c.waitForExpectedData(t, 30*time.Second)
|
|
})
|
|
}
|
|
}
|
|
|
|
func TestShutdown(t *testing.T) {
|
|
deadline := 1 * time.Second
|
|
c := NewTestBlockedWriteClient()
|
|
|
|
cfg := config.DefaultQueueConfig
|
|
mcfg := config.DefaultMetadataConfig
|
|
|
|
m := newTestQueueManager(t, cfg, mcfg, deadline, c, config.RemoteWriteProtoMsgV1)
|
|
n := 2 * config.DefaultQueueConfig.MaxSamplesPerSend
|
|
samples, series := createTimeseries(n, n)
|
|
m.StoreSeries(series, 0)
|
|
m.Start()
|
|
|
|
// Append blocks to guarantee delivery, so we do it in the background.
|
|
go func() {
|
|
m.Append(samples)
|
|
}()
|
|
time.Sleep(100 * time.Millisecond)
|
|
|
|
// Test to ensure that Stop doesn't block.
|
|
start := time.Now()
|
|
m.Stop()
|
|
// The samples will never be delivered, so duration should
|
|
// be at least equal to deadline, otherwise the flush deadline
|
|
// was not respected.
|
|
duration := time.Since(start)
|
|
if duration > deadline+(deadline/10) {
|
|
t.Errorf("Took too long to shutdown: %s > %s", duration, deadline)
|
|
}
|
|
if duration < deadline {
|
|
t.Errorf("Shutdown occurred before flush deadline: %s < %s", duration, deadline)
|
|
}
|
|
}
|
|
|
|
func TestSeriesReset(t *testing.T) {
|
|
c := NewTestBlockedWriteClient()
|
|
deadline := 5 * time.Second
|
|
numSegments := 4
|
|
numSeries := 25
|
|
|
|
cfg := config.DefaultQueueConfig
|
|
mcfg := config.DefaultMetadataConfig
|
|
m := newTestQueueManager(t, cfg, mcfg, deadline, c, config.RemoteWriteProtoMsgV1)
|
|
for i := 0; i < numSegments; i++ {
|
|
series := []record.RefSeries{}
|
|
for j := 0; j < numSeries; j++ {
|
|
series = append(series, record.RefSeries{Ref: chunks.HeadSeriesRef((i * 100) + j), Labels: labels.FromStrings("a", "a")})
|
|
}
|
|
m.StoreSeries(series, i)
|
|
}
|
|
require.Len(t, m.seriesLabels, numSegments*numSeries)
|
|
m.SeriesReset(2)
|
|
require.Len(t, m.seriesLabels, numSegments*numSeries/2)
|
|
}
|
|
|
|
func TestReshard(t *testing.T) {
|
|
for _, protoMsg := range []config.RemoteWriteProtoMsg{config.RemoteWriteProtoMsgV1, config.RemoteWriteProtoMsgV2} {
|
|
t.Run(fmt.Sprint(protoMsg), func(t *testing.T) {
|
|
size := 10 // Make bigger to find more races.
|
|
nSeries := 6
|
|
nSamples := config.DefaultQueueConfig.Capacity * size
|
|
samples, series := createTimeseries(nSamples, nSeries)
|
|
|
|
cfg := config.DefaultQueueConfig
|
|
cfg.MaxShards = 1
|
|
|
|
c := NewTestWriteClient(protoMsg)
|
|
m := newTestQueueManager(t, cfg, config.DefaultMetadataConfig, defaultFlushDeadline, c, protoMsg)
|
|
c.expectSamples(samples, series)
|
|
m.StoreSeries(series, 0)
|
|
|
|
m.Start()
|
|
defer m.Stop()
|
|
|
|
go func() {
|
|
for i := 0; i < len(samples); i += config.DefaultQueueConfig.Capacity {
|
|
sent := m.Append(samples[i : i+config.DefaultQueueConfig.Capacity])
|
|
require.True(t, sent, "samples not sent")
|
|
time.Sleep(100 * time.Millisecond)
|
|
}
|
|
}()
|
|
|
|
for i := 1; i < len(samples)/config.DefaultQueueConfig.Capacity; i++ {
|
|
m.shards.stop()
|
|
m.shards.start(i)
|
|
time.Sleep(100 * time.Millisecond)
|
|
}
|
|
|
|
c.waitForExpectedData(t, 30*time.Second)
|
|
})
|
|
}
|
|
}
|
|
|
|
func TestReshardRaceWithStop(t *testing.T) {
|
|
for _, protoMsg := range []config.RemoteWriteProtoMsg{config.RemoteWriteProtoMsgV1, config.RemoteWriteProtoMsgV2} {
|
|
t.Run(fmt.Sprint(protoMsg), func(t *testing.T) {
|
|
c := NewTestWriteClient(protoMsg)
|
|
var m *QueueManager
|
|
h := sync.Mutex{}
|
|
h.Lock()
|
|
|
|
cfg := testDefaultQueueConfig()
|
|
mcfg := config.DefaultMetadataConfig
|
|
exitCh := make(chan struct{})
|
|
go func() {
|
|
for {
|
|
m = newTestQueueManager(t, cfg, mcfg, defaultFlushDeadline, c, protoMsg)
|
|
|
|
m.Start()
|
|
h.Unlock()
|
|
h.Lock()
|
|
m.Stop()
|
|
|
|
select {
|
|
case exitCh <- struct{}{}:
|
|
return
|
|
default:
|
|
}
|
|
}
|
|
}()
|
|
|
|
for i := 1; i < 100; i++ {
|
|
h.Lock()
|
|
m.reshardChan <- i
|
|
h.Unlock()
|
|
}
|
|
<-exitCh
|
|
})
|
|
}
|
|
}
|
|
|
|
func TestReshardPartialBatch(t *testing.T) {
|
|
for _, protoMsg := range []config.RemoteWriteProtoMsg{config.RemoteWriteProtoMsgV1, config.RemoteWriteProtoMsgV2} {
|
|
t.Run(fmt.Sprint(protoMsg), func(t *testing.T) {
|
|
samples, series := createTimeseries(1, 10)
|
|
|
|
c := NewTestBlockedWriteClient()
|
|
|
|
cfg := testDefaultQueueConfig()
|
|
mcfg := config.DefaultMetadataConfig
|
|
cfg.MaxShards = 1
|
|
batchSendDeadline := time.Millisecond
|
|
flushDeadline := 10 * time.Millisecond
|
|
cfg.BatchSendDeadline = model.Duration(batchSendDeadline)
|
|
|
|
m := newTestQueueManager(t, cfg, mcfg, flushDeadline, c, protoMsg)
|
|
m.StoreSeries(series, 0)
|
|
|
|
m.Start()
|
|
|
|
for i := 0; i < 100; i++ {
|
|
done := make(chan struct{})
|
|
go func() {
|
|
m.Append(samples)
|
|
time.Sleep(batchSendDeadline)
|
|
m.shards.stop()
|
|
m.shards.start(1)
|
|
done <- struct{}{}
|
|
}()
|
|
select {
|
|
case <-done:
|
|
case <-time.After(2 * time.Second):
|
|
t.Error("Deadlock between sending and stopping detected")
|
|
pprof.Lookup("goroutine").WriteTo(os.Stdout, 1)
|
|
t.FailNow()
|
|
}
|
|
}
|
|
// We can only call stop if there was not a deadlock.
|
|
m.Stop()
|
|
})
|
|
}
|
|
}
|
|
|
|
// TestQueueFilledDeadlock makes sure the code does not deadlock in the case
|
|
// where a large scrape (> capacity + max samples per send) is appended at the
|
|
// same time as a batch times out according to the batch send deadline.
|
|
func TestQueueFilledDeadlock(t *testing.T) {
|
|
for _, protoMsg := range []config.RemoteWriteProtoMsg{config.RemoteWriteProtoMsgV1, config.RemoteWriteProtoMsgV2} {
|
|
t.Run(fmt.Sprint(protoMsg), func(t *testing.T) {
|
|
samples, series := createTimeseries(50, 1)
|
|
|
|
c := NewNopWriteClient()
|
|
|
|
cfg := testDefaultQueueConfig()
|
|
mcfg := config.DefaultMetadataConfig
|
|
cfg.MaxShards = 1
|
|
cfg.MaxSamplesPerSend = 10
|
|
cfg.Capacity = 20
|
|
flushDeadline := time.Second
|
|
batchSendDeadline := time.Millisecond
|
|
cfg.BatchSendDeadline = model.Duration(batchSendDeadline)
|
|
|
|
m := newTestQueueManager(t, cfg, mcfg, flushDeadline, c, protoMsg)
|
|
m.StoreSeries(series, 0)
|
|
m.Start()
|
|
defer m.Stop()
|
|
|
|
for i := 0; i < 100; i++ {
|
|
done := make(chan struct{})
|
|
go func() {
|
|
time.Sleep(batchSendDeadline)
|
|
m.Append(samples)
|
|
done <- struct{}{}
|
|
}()
|
|
select {
|
|
case <-done:
|
|
case <-time.After(2 * time.Second):
|
|
t.Error("Deadlock between sending and appending detected")
|
|
pprof.Lookup("goroutine").WriteTo(os.Stdout, 1)
|
|
t.FailNow()
|
|
}
|
|
}
|
|
})
|
|
}
|
|
}
|
|
|
|
func TestReleaseNoninternedString(t *testing.T) {
|
|
for _, protoMsg := range []config.RemoteWriteProtoMsg{config.RemoteWriteProtoMsgV1, config.RemoteWriteProtoMsgV2} {
|
|
t.Run(fmt.Sprint(protoMsg), func(t *testing.T) {
|
|
_, m := newTestClientAndQueueManager(t, defaultFlushDeadline, protoMsg)
|
|
m.Start()
|
|
defer m.Stop()
|
|
for i := 1; i < 1000; i++ {
|
|
m.StoreSeries([]record.RefSeries{
|
|
{
|
|
Ref: chunks.HeadSeriesRef(i),
|
|
Labels: labels.FromStrings("asdf", strconv.Itoa(i)),
|
|
},
|
|
}, 0)
|
|
m.SeriesReset(1)
|
|
}
|
|
|
|
metric := client_testutil.ToFloat64(noReferenceReleases)
|
|
require.Equal(t, 0.0, metric, "expected there to be no calls to release for strings that were not already interned: %d", int(metric))
|
|
})
|
|
}
|
|
}
|
|
|
|
func TestShouldReshard(t *testing.T) {
|
|
type testcase struct {
|
|
startingShards int
|
|
samplesIn, samplesOut, lastSendTimestamp int64
|
|
expectedToReshard bool
|
|
sendDeadline model.Duration
|
|
}
|
|
cases := []testcase{
|
|
{
|
|
// resharding shouldn't take place if we haven't successfully sent
|
|
// since the last shardUpdateDuration, even if the send deadline is very low
|
|
startingShards: 10,
|
|
samplesIn: 1000,
|
|
samplesOut: 10,
|
|
lastSendTimestamp: time.Now().Unix() - int64(shardUpdateDuration),
|
|
expectedToReshard: false,
|
|
sendDeadline: model.Duration(100 * time.Millisecond),
|
|
},
|
|
{
|
|
startingShards: 10,
|
|
samplesIn: 1000,
|
|
samplesOut: 10,
|
|
lastSendTimestamp: time.Now().Unix(),
|
|
expectedToReshard: true,
|
|
sendDeadline: config.DefaultQueueConfig.BatchSendDeadline,
|
|
},
|
|
}
|
|
|
|
for _, c := range cases {
|
|
_, m := newTestClientAndQueueManager(t, time.Duration(c.sendDeadline), config.RemoteWriteProtoMsgV1)
|
|
m.numShards = c.startingShards
|
|
m.dataIn.incr(c.samplesIn)
|
|
m.dataOut.incr(c.samplesOut)
|
|
m.lastSendTimestamp.Store(c.lastSendTimestamp)
|
|
m.Start()
|
|
|
|
desiredShards := m.calculateDesiredShards()
|
|
shouldReshard := m.shouldReshard(desiredShards)
|
|
|
|
m.Stop()
|
|
|
|
require.Equal(t, c.expectedToReshard, shouldReshard)
|
|
}
|
|
}
|
|
|
|
// TestDisableReshardOnRetry asserts that resharding should be disabled when a
|
|
// recoverable error is returned from remote_write.
|
|
func TestDisableReshardOnRetry(t *testing.T) {
|
|
onStoredContext, onStoreCalled := context.WithCancel(context.Background())
|
|
defer onStoreCalled()
|
|
|
|
var (
|
|
fakeSamples, fakeSeries = createTimeseries(100, 100)
|
|
|
|
cfg = config.DefaultQueueConfig
|
|
mcfg = config.DefaultMetadataConfig
|
|
retryAfter = time.Second
|
|
|
|
metrics = newQueueManagerMetrics(nil, "", "")
|
|
|
|
client = &MockWriteClient{
|
|
StoreFunc: func(ctx context.Context, b []byte, i int) (WriteResponseStats, error) {
|
|
onStoreCalled()
|
|
|
|
return WriteResponseStats{}, RecoverableError{
|
|
error: fmt.Errorf("fake error"),
|
|
retryAfter: model.Duration(retryAfter),
|
|
}
|
|
},
|
|
NameFunc: func() string { return "mock" },
|
|
EndpointFunc: func() string { return "http://fake:9090/api/v1/write" },
|
|
}
|
|
)
|
|
|
|
m := NewQueueManager(metrics, nil, nil, nil, "", newEWMARate(ewmaWeight, shardUpdateDuration), cfg, mcfg, labels.EmptyLabels(), nil, client, 0, newPool(), newHighestTimestampMetric(), nil, false, false, config.RemoteWriteProtoMsgV1)
|
|
m.StoreSeries(fakeSeries, 0)
|
|
|
|
// Attempt to samples while the manager is running. We immediately stop the
|
|
// manager after the recoverable error is generated to prevent the manager
|
|
// from resharding itself.
|
|
m.Start()
|
|
{
|
|
m.Append(fakeSamples)
|
|
|
|
select {
|
|
case <-onStoredContext.Done():
|
|
case <-time.After(time.Minute):
|
|
require.FailNow(t, "timed out waiting for client to be sent metrics")
|
|
}
|
|
}
|
|
m.Stop()
|
|
|
|
require.Eventually(t, func() bool {
|
|
// Force m.lastSendTimestamp to be current so the last send timestamp isn't
|
|
// the reason resharding is disabled.
|
|
m.lastSendTimestamp.Store(time.Now().Unix())
|
|
return m.shouldReshard(m.numShards+1) == false
|
|
}, time.Minute, 10*time.Millisecond, "shouldReshard was never disabled")
|
|
|
|
// After 2x retryAfter, resharding should be enabled again.
|
|
require.Eventually(t, func() bool {
|
|
// Force m.lastSendTimestamp to be current so the last send timestamp isn't
|
|
// the reason resharding is disabled.
|
|
m.lastSendTimestamp.Store(time.Now().Unix())
|
|
return m.shouldReshard(m.numShards+1) == true
|
|
}, time.Minute, retryAfter, "shouldReshard should have been re-enabled")
|
|
}
|
|
|
|
func createTimeseries(numSamples, numSeries int, extraLabels ...labels.Label) ([]record.RefSample, []record.RefSeries) {
|
|
samples := make([]record.RefSample, 0, numSamples)
|
|
series := make([]record.RefSeries, 0, numSeries)
|
|
lb := labels.NewScratchBuilder(1 + len(extraLabels))
|
|
for i := 0; i < numSeries; i++ {
|
|
name := fmt.Sprintf("test_metric_%d", i)
|
|
for j := 0; j < numSamples; j++ {
|
|
samples = append(samples, record.RefSample{
|
|
Ref: chunks.HeadSeriesRef(i),
|
|
T: int64(j),
|
|
V: float64(i),
|
|
})
|
|
}
|
|
// Create Labels that is name of series plus any extra labels supplied.
|
|
lb.Reset()
|
|
lb.Add(labels.MetricName, name)
|
|
rand.Shuffle(len(extraLabels), func(i, j int) {
|
|
extraLabels[i], extraLabels[j] = extraLabels[j], extraLabels[i]
|
|
})
|
|
for _, l := range extraLabels {
|
|
lb.Add(l.Name, l.Value)
|
|
}
|
|
lb.Sort()
|
|
series = append(series, record.RefSeries{
|
|
Ref: chunks.HeadSeriesRef(i),
|
|
Labels: lb.Labels(),
|
|
})
|
|
}
|
|
return samples, series
|
|
}
|
|
|
|
func createProtoTimeseriesWithOld(numSamples, baseTs int64, extraLabels ...labels.Label) []prompb.TimeSeries {
|
|
samples := make([]prompb.TimeSeries, numSamples)
|
|
// use a fixed rand source so tests are consistent
|
|
r := rand.New(rand.NewSource(99))
|
|
for j := int64(0); j < numSamples; j++ {
|
|
name := fmt.Sprintf("test_metric_%d", j)
|
|
|
|
samples[j] = prompb.TimeSeries{
|
|
Labels: []prompb.Label{{Name: "__name__", Value: name}},
|
|
Samples: []prompb.Sample{
|
|
{
|
|
Timestamp: baseTs + j,
|
|
Value: float64(j),
|
|
},
|
|
},
|
|
}
|
|
// 10% of the time use a ts that is too old
|
|
if r.Intn(10) == 0 {
|
|
samples[j].Samples[0].Timestamp = baseTs - 5
|
|
}
|
|
}
|
|
return samples
|
|
}
|
|
|
|
func createExemplars(numExemplars, numSeries int) ([]record.RefExemplar, []record.RefSeries) {
|
|
exemplars := make([]record.RefExemplar, 0, numExemplars)
|
|
series := make([]record.RefSeries, 0, numSeries)
|
|
for i := 0; i < numSeries; i++ {
|
|
name := fmt.Sprintf("test_metric_%d", i)
|
|
for j := 0; j < numExemplars; j++ {
|
|
e := record.RefExemplar{
|
|
Ref: chunks.HeadSeriesRef(i),
|
|
T: int64(j),
|
|
V: float64(i),
|
|
Labels: labels.FromStrings("trace_id", fmt.Sprintf("trace-%d", i)),
|
|
}
|
|
exemplars = append(exemplars, e)
|
|
}
|
|
series = append(series, record.RefSeries{
|
|
Ref: chunks.HeadSeriesRef(i),
|
|
Labels: labels.FromStrings("__name__", name),
|
|
})
|
|
}
|
|
return exemplars, series
|
|
}
|
|
|
|
func createHistograms(numSamples, numSeries int, floatHistogram bool) ([]record.RefHistogramSample, []record.RefFloatHistogramSample, []record.RefSeries) {
|
|
histograms := make([]record.RefHistogramSample, 0, numSamples)
|
|
floatHistograms := make([]record.RefFloatHistogramSample, 0, numSamples)
|
|
series := make([]record.RefSeries, 0, numSeries)
|
|
for i := 0; i < numSeries; i++ {
|
|
name := fmt.Sprintf("test_metric_%d", i)
|
|
for j := 0; j < numSamples; j++ {
|
|
hist := &histogram.Histogram{
|
|
Schema: 2,
|
|
ZeroThreshold: 1e-128,
|
|
ZeroCount: 0,
|
|
Count: 2,
|
|
Sum: 0,
|
|
PositiveSpans: []histogram.Span{{Offset: 0, Length: 1}},
|
|
PositiveBuckets: []int64{int64(i) + 1},
|
|
NegativeSpans: []histogram.Span{{Offset: 0, Length: 1}},
|
|
NegativeBuckets: []int64{int64(-i) - 1},
|
|
}
|
|
|
|
if floatHistogram {
|
|
fh := record.RefFloatHistogramSample{
|
|
Ref: chunks.HeadSeriesRef(i),
|
|
T: int64(j),
|
|
FH: hist.ToFloat(nil),
|
|
}
|
|
floatHistograms = append(floatHistograms, fh)
|
|
} else {
|
|
h := record.RefHistogramSample{
|
|
Ref: chunks.HeadSeriesRef(i),
|
|
T: int64(j),
|
|
H: hist,
|
|
}
|
|
histograms = append(histograms, h)
|
|
}
|
|
}
|
|
series = append(series, record.RefSeries{
|
|
Ref: chunks.HeadSeriesRef(i),
|
|
Labels: labels.FromStrings("__name__", name),
|
|
})
|
|
}
|
|
if floatHistogram {
|
|
return nil, floatHistograms, series
|
|
}
|
|
return histograms, nil, series
|
|
}
|
|
|
|
func createSeriesMetadata(series []record.RefSeries) []record.RefMetadata {
|
|
metas := make([]record.RefMetadata, 0, len(series))
|
|
|
|
for _, s := range series {
|
|
metas = append(metas, record.RefMetadata{
|
|
Ref: s.Ref,
|
|
Type: uint8(record.Counter),
|
|
Unit: "unit text",
|
|
Help: "help text",
|
|
})
|
|
}
|
|
return metas
|
|
}
|
|
|
|
func getSeriesIDFromRef(r record.RefSeries) string {
|
|
return r.Labels.String()
|
|
}
|
|
|
|
// TestWriteClient represents write client which does not call remote storage,
|
|
// but instead re-implements fake WriteHandler for test purposes.
|
|
type TestWriteClient struct {
|
|
receivedSamples map[string][]prompb.Sample
|
|
expectedSamples map[string][]prompb.Sample
|
|
receivedExemplars map[string][]prompb.Exemplar
|
|
expectedExemplars map[string][]prompb.Exemplar
|
|
receivedHistograms map[string][]prompb.Histogram
|
|
receivedFloatHistograms map[string][]prompb.Histogram
|
|
expectedHistograms map[string][]prompb.Histogram
|
|
expectedFloatHistograms map[string][]prompb.Histogram
|
|
receivedMetadata map[string][]prompb.MetricMetadata
|
|
writesReceived int
|
|
mtx sync.Mutex
|
|
buf []byte
|
|
protoMsg config.RemoteWriteProtoMsg
|
|
injectedErrs []error
|
|
currErr int
|
|
retry bool
|
|
|
|
storeWait time.Duration
|
|
// TODO(npazosmendez): maybe replaceable with injectedErrs?
|
|
returnError error
|
|
}
|
|
|
|
// NewTestWriteClient creates a new testing write client.
|
|
func NewTestWriteClient(protoMsg config.RemoteWriteProtoMsg) *TestWriteClient {
|
|
return &TestWriteClient{
|
|
receivedSamples: map[string][]prompb.Sample{},
|
|
expectedSamples: map[string][]prompb.Sample{},
|
|
receivedMetadata: map[string][]prompb.MetricMetadata{},
|
|
protoMsg: protoMsg,
|
|
storeWait: 0,
|
|
returnError: nil,
|
|
}
|
|
}
|
|
|
|
func (c *TestWriteClient) injectErrors(injectedErrs []error) {
|
|
c.injectedErrs = injectedErrs
|
|
c.currErr = -1
|
|
c.retry = false
|
|
}
|
|
|
|
func (c *TestWriteClient) expectSamples(ss []record.RefSample, series []record.RefSeries) {
|
|
c.mtx.Lock()
|
|
defer c.mtx.Unlock()
|
|
|
|
c.expectedSamples = map[string][]prompb.Sample{}
|
|
c.receivedSamples = map[string][]prompb.Sample{}
|
|
|
|
for _, s := range ss {
|
|
tsID := getSeriesIDFromRef(series[s.Ref])
|
|
c.expectedSamples[tsID] = append(c.expectedSamples[tsID], prompb.Sample{
|
|
Timestamp: s.T,
|
|
Value: s.V,
|
|
})
|
|
}
|
|
}
|
|
|
|
func (c *TestWriteClient) expectExemplars(ss []record.RefExemplar, series []record.RefSeries) {
|
|
c.mtx.Lock()
|
|
defer c.mtx.Unlock()
|
|
|
|
c.expectedExemplars = map[string][]prompb.Exemplar{}
|
|
c.receivedExemplars = map[string][]prompb.Exemplar{}
|
|
|
|
for _, s := range ss {
|
|
tsID := getSeriesIDFromRef(series[s.Ref])
|
|
e := prompb.Exemplar{
|
|
Labels: prompb.FromLabels(s.Labels, nil),
|
|
Timestamp: s.T,
|
|
Value: s.V,
|
|
}
|
|
c.expectedExemplars[tsID] = append(c.expectedExemplars[tsID], e)
|
|
}
|
|
}
|
|
|
|
func (c *TestWriteClient) expectHistograms(hh []record.RefHistogramSample, series []record.RefSeries) {
|
|
c.mtx.Lock()
|
|
defer c.mtx.Unlock()
|
|
|
|
c.expectedHistograms = map[string][]prompb.Histogram{}
|
|
c.receivedHistograms = map[string][]prompb.Histogram{}
|
|
|
|
for _, h := range hh {
|
|
tsID := getSeriesIDFromRef(series[h.Ref])
|
|
c.expectedHistograms[tsID] = append(c.expectedHistograms[tsID], prompb.FromIntHistogram(h.T, h.H))
|
|
}
|
|
}
|
|
|
|
func (c *TestWriteClient) expectFloatHistograms(fhs []record.RefFloatHistogramSample, series []record.RefSeries) {
|
|
c.mtx.Lock()
|
|
defer c.mtx.Unlock()
|
|
|
|
c.expectedFloatHistograms = map[string][]prompb.Histogram{}
|
|
c.receivedFloatHistograms = map[string][]prompb.Histogram{}
|
|
|
|
for _, fh := range fhs {
|
|
tsID := getSeriesIDFromRef(series[fh.Ref])
|
|
c.expectedFloatHistograms[tsID] = append(c.expectedFloatHistograms[tsID], prompb.FromFloatHistogram(fh.T, fh.FH))
|
|
}
|
|
}
|
|
|
|
func deepLen[M any](ms ...map[string][]M) int {
|
|
l := 0
|
|
for _, m := range ms {
|
|
for _, v := range m {
|
|
l += len(v)
|
|
}
|
|
}
|
|
return l
|
|
}
|
|
|
|
func (c *TestWriteClient) waitForExpectedData(tb testing.TB, timeout time.Duration) {
|
|
tb.Helper()
|
|
|
|
ctx, cancel := context.WithTimeout(context.Background(), timeout)
|
|
defer cancel()
|
|
if err := runutil.Retry(500*time.Millisecond, ctx.Done(), func() error {
|
|
c.mtx.Lock()
|
|
exp := deepLen(c.expectedSamples) + deepLen(c.expectedExemplars) + deepLen(c.expectedHistograms, c.expectedFloatHistograms)
|
|
got := deepLen(c.receivedSamples) + deepLen(c.receivedExemplars) + deepLen(c.receivedHistograms, c.receivedFloatHistograms)
|
|
c.mtx.Unlock()
|
|
|
|
if got < exp {
|
|
return fmt.Errorf("expected %v samples/exemplars/histograms/floathistograms, got %v", exp, got)
|
|
}
|
|
return nil
|
|
}); err != nil {
|
|
tb.Error(err)
|
|
}
|
|
|
|
c.mtx.Lock()
|
|
defer c.mtx.Unlock()
|
|
|
|
for ts, expectedSamples := range c.expectedSamples {
|
|
require.Equal(tb, expectedSamples, c.receivedSamples[ts], ts)
|
|
}
|
|
for ts, expectedExemplar := range c.expectedExemplars {
|
|
require.Equal(tb, expectedExemplar, c.receivedExemplars[ts], ts)
|
|
}
|
|
for ts, expectedHistogram := range c.expectedHistograms {
|
|
require.Equal(tb, expectedHistogram, c.receivedHistograms[ts], ts)
|
|
}
|
|
for ts, expectedFloatHistogram := range c.expectedFloatHistograms {
|
|
require.Equal(tb, expectedFloatHistogram, c.receivedFloatHistograms[ts], ts)
|
|
}
|
|
}
|
|
|
|
func (c *TestWriteClient) SetStoreWait(w time.Duration) {
|
|
c.mtx.Lock()
|
|
defer c.mtx.Unlock()
|
|
c.storeWait = w
|
|
}
|
|
|
|
func (c *TestWriteClient) SetReturnError(err error) {
|
|
c.mtx.Lock()
|
|
defer c.mtx.Unlock()
|
|
c.returnError = err
|
|
}
|
|
|
|
func (c *TestWriteClient) Store(_ context.Context, req []byte, _ int) (WriteResponseStats, error) {
|
|
c.mtx.Lock()
|
|
defer c.mtx.Unlock()
|
|
if c.storeWait > 0 {
|
|
time.Sleep(c.storeWait)
|
|
}
|
|
if c.returnError != nil {
|
|
return WriteResponseStats{}, c.returnError
|
|
}
|
|
// nil buffers are ok for snappy, ignore cast error.
|
|
if c.buf != nil {
|
|
c.buf = c.buf[:cap(c.buf)]
|
|
}
|
|
|
|
reqBuf, err := snappy.Decode(c.buf, req)
|
|
c.buf = reqBuf
|
|
if err != nil {
|
|
return WriteResponseStats{}, err
|
|
}
|
|
|
|
// Check if we've been told to inject err for this call.
|
|
if len(c.injectedErrs) > 0 {
|
|
c.currErr++
|
|
if err = c.injectedErrs[c.currErr]; err != nil {
|
|
return WriteResponseStats{}, err
|
|
}
|
|
}
|
|
|
|
var reqProto *prompb.WriteRequest
|
|
switch c.protoMsg {
|
|
case config.RemoteWriteProtoMsgV1:
|
|
reqProto = &prompb.WriteRequest{}
|
|
err = proto.Unmarshal(reqBuf, reqProto)
|
|
case config.RemoteWriteProtoMsgV2:
|
|
// NOTE(bwplotka): v1 msg can be unmarshaled to v2 sometimes, without
|
|
// errors.
|
|
var reqProtoV2 writev2.Request
|
|
err = proto.Unmarshal(reqBuf, &reqProtoV2)
|
|
if err == nil {
|
|
reqProto, err = v2RequestToWriteRequest(&reqProtoV2)
|
|
}
|
|
}
|
|
if err != nil {
|
|
return WriteResponseStats{}, err
|
|
}
|
|
|
|
rs := WriteResponseStats{}
|
|
b := labels.NewScratchBuilder(0)
|
|
for _, ts := range reqProto.Timeseries {
|
|
labels := ts.ToLabels(&b, nil)
|
|
tsID := labels.String()
|
|
if len(ts.Samples) > 0 {
|
|
c.receivedSamples[tsID] = append(c.receivedSamples[tsID], ts.Samples...)
|
|
}
|
|
rs.Samples += len(ts.Samples)
|
|
|
|
if len(ts.Exemplars) > 0 {
|
|
c.receivedExemplars[tsID] = append(c.receivedExemplars[tsID], ts.Exemplars...)
|
|
}
|
|
rs.Exemplars += len(ts.Exemplars)
|
|
|
|
for _, h := range ts.Histograms {
|
|
if h.IsFloatHistogram() {
|
|
c.receivedFloatHistograms[tsID] = append(c.receivedFloatHistograms[tsID], h)
|
|
} else {
|
|
c.receivedHistograms[tsID] = append(c.receivedHistograms[tsID], h)
|
|
}
|
|
}
|
|
rs.Histograms += len(ts.Histograms)
|
|
}
|
|
for _, m := range reqProto.Metadata {
|
|
c.receivedMetadata[m.MetricFamilyName] = append(c.receivedMetadata[m.MetricFamilyName], m)
|
|
}
|
|
|
|
c.writesReceived++
|
|
return rs, nil
|
|
}
|
|
|
|
func (c *TestWriteClient) Name() string {
|
|
return "testwriteclient"
|
|
}
|
|
|
|
func (c *TestWriteClient) Endpoint() string {
|
|
return "http://test-remote.com/1234"
|
|
}
|
|
|
|
func v2RequestToWriteRequest(v2Req *writev2.Request) (*prompb.WriteRequest, error) {
|
|
req := &prompb.WriteRequest{
|
|
Timeseries: make([]prompb.TimeSeries, len(v2Req.Timeseries)),
|
|
// TODO handle metadata?
|
|
}
|
|
b := labels.NewScratchBuilder(0)
|
|
for i, rts := range v2Req.Timeseries {
|
|
rts.ToLabels(&b, v2Req.Symbols).Range(func(l labels.Label) {
|
|
req.Timeseries[i].Labels = append(req.Timeseries[i].Labels, prompb.Label{
|
|
Name: l.Name,
|
|
Value: l.Value,
|
|
})
|
|
})
|
|
|
|
exemplars := make([]prompb.Exemplar, len(rts.Exemplars))
|
|
for j, e := range rts.Exemplars {
|
|
exemplars[j].Value = e.Value
|
|
exemplars[j].Timestamp = e.Timestamp
|
|
e.ToExemplar(&b, v2Req.Symbols).Labels.Range(func(l labels.Label) {
|
|
exemplars[j].Labels = append(exemplars[j].Labels, prompb.Label{
|
|
Name: l.Name,
|
|
Value: l.Value,
|
|
})
|
|
})
|
|
}
|
|
req.Timeseries[i].Exemplars = exemplars
|
|
|
|
req.Timeseries[i].Samples = make([]prompb.Sample, len(rts.Samples))
|
|
for j, s := range rts.Samples {
|
|
req.Timeseries[i].Samples[j].Timestamp = s.Timestamp
|
|
req.Timeseries[i].Samples[j].Value = s.Value
|
|
}
|
|
|
|
req.Timeseries[i].Histograms = make([]prompb.Histogram, len(rts.Histograms))
|
|
for j, h := range rts.Histograms {
|
|
if h.IsFloatHistogram() {
|
|
req.Timeseries[i].Histograms[j] = prompb.FromFloatHistogram(h.Timestamp, h.ToFloatHistogram())
|
|
continue
|
|
}
|
|
req.Timeseries[i].Histograms[j] = prompb.FromIntHistogram(h.Timestamp, h.ToIntHistogram())
|
|
}
|
|
}
|
|
return req, nil
|
|
}
|
|
|
|
// TestBlockingWriteClient is a queue_manager WriteClient which will block
|
|
// on any calls to Store(), until the request's Context is cancelled, at which
|
|
// point the `numCalls` property will contain a count of how many times Store()
|
|
// was called.
|
|
type TestBlockingWriteClient struct {
|
|
numCalls atomic.Uint64
|
|
}
|
|
|
|
func NewTestBlockedWriteClient() *TestBlockingWriteClient {
|
|
return &TestBlockingWriteClient{}
|
|
}
|
|
|
|
func (c *TestBlockingWriteClient) Store(ctx context.Context, _ []byte, _ int) (WriteResponseStats, error) {
|
|
c.numCalls.Inc()
|
|
<-ctx.Done()
|
|
return WriteResponseStats{}, nil
|
|
}
|
|
|
|
func (c *TestBlockingWriteClient) NumCalls() uint64 {
|
|
return c.numCalls.Load()
|
|
}
|
|
|
|
func (c *TestBlockingWriteClient) Name() string {
|
|
return "testblockingwriteclient"
|
|
}
|
|
|
|
func (c *TestBlockingWriteClient) Endpoint() string {
|
|
return "http://test-remote-blocking.com/1234"
|
|
}
|
|
|
|
// For benchmarking the send and not the receive side.
|
|
type NopWriteClient struct{}
|
|
|
|
func NewNopWriteClient() *NopWriteClient { return &NopWriteClient{} }
|
|
func (c *NopWriteClient) Store(context.Context, []byte, int) (WriteResponseStats, error) {
|
|
return WriteResponseStats{}, nil
|
|
}
|
|
func (c *NopWriteClient) Name() string { return "nopwriteclient" }
|
|
func (c *NopWriteClient) Endpoint() string { return "http://test-remote.com/1234" }
|
|
|
|
type MockWriteClient struct {
|
|
StoreFunc func(context.Context, []byte, int) (WriteResponseStats, error)
|
|
NameFunc func() string
|
|
EndpointFunc func() string
|
|
}
|
|
|
|
func (c *MockWriteClient) Store(ctx context.Context, bb []byte, n int) (WriteResponseStats, error) {
|
|
return c.StoreFunc(ctx, bb, n)
|
|
}
|
|
func (c *MockWriteClient) Name() string { return c.NameFunc() }
|
|
func (c *MockWriteClient) Endpoint() string { return c.EndpointFunc() }
|
|
|
|
// Extra labels to make a more realistic workload - taken from Kubernetes' embedded cAdvisor metrics.
|
|
var extraLabels []labels.Label = []labels.Label{
|
|
{Name: "kubernetes_io_arch", Value: "amd64"},
|
|
{Name: "kubernetes_io_instance_type", Value: "c3.somesize"},
|
|
{Name: "kubernetes_io_os", Value: "linux"},
|
|
{Name: "container_name", Value: "some-name"},
|
|
{Name: "failure_domain_kubernetes_io_region", Value: "somewhere-1"},
|
|
{Name: "failure_domain_kubernetes_io_zone", Value: "somewhere-1b"},
|
|
{Name: "id", Value: "/kubepods/burstable/pod6e91c467-e4c5-11e7-ace3-0a97ed59c75e/a3c8498918bd6866349fed5a6f8c643b77c91836427fb6327913276ebc6bde28"},
|
|
{Name: "image", Value: "registry/organisation/name@sha256:dca3d877a80008b45d71d7edc4fd2e44c0c8c8e7102ba5cbabec63a374d1d506"},
|
|
{Name: "instance", Value: "ip-111-11-1-11.ec2.internal"},
|
|
{Name: "job", Value: "kubernetes-cadvisor"},
|
|
{Name: "kubernetes_io_hostname", Value: "ip-111-11-1-11"},
|
|
{Name: "monitor", Value: "prod"},
|
|
{Name: "name", Value: "k8s_some-name_some-other-name-5j8s8_kube-system_6e91c467-e4c5-11e7-ace3-0a97ed59c75e_0"},
|
|
{Name: "namespace", Value: "kube-system"},
|
|
{Name: "pod_name", Value: "some-other-name-5j8s8"},
|
|
}
|
|
|
|
func BenchmarkSampleSend(b *testing.B) {
|
|
// Send one sample per series, which is the typical remote_write case
|
|
const numSamples = 1
|
|
const numSeries = 10000
|
|
|
|
samples, series := createTimeseries(numSamples, numSeries, extraLabels...)
|
|
|
|
c := NewNopWriteClient()
|
|
|
|
cfg := testDefaultQueueConfig()
|
|
mcfg := config.DefaultMetadataConfig
|
|
cfg.BatchSendDeadline = model.Duration(100 * time.Millisecond)
|
|
cfg.MinShards = 20
|
|
cfg.MaxShards = 20
|
|
|
|
// todo: test with new proto type(s)
|
|
m := newTestQueueManager(b, cfg, mcfg, defaultFlushDeadline, c, config.RemoteWriteProtoMsgV1)
|
|
m.StoreSeries(series, 0)
|
|
|
|
// These should be received by the client.
|
|
m.Start()
|
|
defer m.Stop()
|
|
|
|
b.ResetTimer()
|
|
for i := 0; i < b.N; i++ {
|
|
m.Append(samples)
|
|
m.UpdateSeriesSegment(series, i+1) // simulate what wlog.Watcher.garbageCollectSeries does
|
|
m.SeriesReset(i + 1)
|
|
}
|
|
// Do not include shutdown
|
|
b.StopTimer()
|
|
}
|
|
|
|
// Check how long it takes to add N series, including external labels processing.
|
|
func BenchmarkStoreSeries(b *testing.B) {
|
|
externalLabels := []labels.Label{
|
|
{Name: "cluster", Value: "mycluster"},
|
|
{Name: "replica", Value: "1"},
|
|
}
|
|
relabelConfigs := []*relabel.Config{{
|
|
SourceLabels: model.LabelNames{"namespace"},
|
|
Separator: ";",
|
|
Regex: relabel.MustNewRegexp("kube.*"),
|
|
TargetLabel: "job",
|
|
Replacement: "$1",
|
|
Action: relabel.Replace,
|
|
}}
|
|
testCases := []struct {
|
|
name string
|
|
externalLabels []labels.Label
|
|
ts []prompb.TimeSeries
|
|
relabelConfigs []*relabel.Config
|
|
}{
|
|
{name: "plain"},
|
|
{name: "externalLabels", externalLabels: externalLabels},
|
|
{name: "relabel", relabelConfigs: relabelConfigs},
|
|
{
|
|
name: "externalLabels+relabel",
|
|
externalLabels: externalLabels,
|
|
relabelConfigs: relabelConfigs,
|
|
},
|
|
}
|
|
|
|
// numSeries chosen to be big enough that StoreSeries dominates creating a new queue manager.
|
|
const numSeries = 1000
|
|
_, series := createTimeseries(0, numSeries, extraLabels...)
|
|
|
|
for _, tc := range testCases {
|
|
b.Run(tc.name, func(b *testing.B) {
|
|
for i := 0; i < b.N; i++ {
|
|
c := NewTestWriteClient(config.RemoteWriteProtoMsgV1)
|
|
dir := b.TempDir()
|
|
cfg := config.DefaultQueueConfig
|
|
mcfg := config.DefaultMetadataConfig
|
|
metrics := newQueueManagerMetrics(nil, "", "")
|
|
m := NewQueueManager(metrics, nil, nil, nil, dir, newEWMARate(ewmaWeight, shardUpdateDuration), cfg, mcfg, labels.EmptyLabels(), nil, c, defaultFlushDeadline, newPool(), newHighestTimestampMetric(), nil, false, false, config.RemoteWriteProtoMsgV1)
|
|
m.externalLabels = tc.externalLabels
|
|
m.relabelConfigs = tc.relabelConfigs
|
|
|
|
m.StoreSeries(series, 0)
|
|
}
|
|
})
|
|
}
|
|
}
|
|
|
|
func BenchmarkStartup(b *testing.B) {
|
|
dir := os.Getenv("WALDIR")
|
|
if dir == "" {
|
|
b.Skip("WALDIR env var not set")
|
|
}
|
|
|
|
// Find the second largest segment; we will replay up to this.
|
|
// (Second largest as WALWatcher will start tailing the largest).
|
|
dirents, err := os.ReadDir(dir)
|
|
require.NoError(b, err)
|
|
|
|
var segments []int
|
|
for _, dirent := range dirents {
|
|
if i, err := strconv.Atoi(dirent.Name()); err != nil {
|
|
segments = append(segments, i)
|
|
}
|
|
}
|
|
sort.Ints(segments)
|
|
|
|
logger := log.NewLogfmtLogger(log.NewSyncWriter(os.Stdout))
|
|
logger = log.With(logger, "caller", log.DefaultCaller)
|
|
|
|
cfg := testDefaultQueueConfig()
|
|
mcfg := config.DefaultMetadataConfig
|
|
for n := 0; n < b.N; n++ {
|
|
metrics := newQueueManagerMetrics(nil, "", "")
|
|
c := NewTestBlockedWriteClient()
|
|
// todo: test with new proto type(s)
|
|
m := NewQueueManager(metrics, nil, nil, logger, dir,
|
|
newEWMARate(ewmaWeight, shardUpdateDuration),
|
|
cfg, mcfg, labels.EmptyLabels(), nil, c, 1*time.Minute, newPool(), newHighestTimestampMetric(), nil, false, false, config.RemoteWriteProtoMsgV1)
|
|
m.watcher.SetStartTime(timestamp.Time(math.MaxInt64))
|
|
m.watcher.MaxSegment = segments[len(segments)-2]
|
|
err := m.watcher.Run()
|
|
require.NoError(b, err)
|
|
}
|
|
}
|
|
|
|
func TestProcessExternalLabels(t *testing.T) {
|
|
b := labels.NewBuilder(labels.EmptyLabels())
|
|
for i, tc := range []struct {
|
|
labels labels.Labels
|
|
externalLabels []labels.Label
|
|
expected labels.Labels
|
|
}{
|
|
// Test adding labels at the end.
|
|
{
|
|
labels: labels.FromStrings("a", "b"),
|
|
externalLabels: []labels.Label{{Name: "c", Value: "d"}},
|
|
expected: labels.FromStrings("a", "b", "c", "d"),
|
|
},
|
|
|
|
// Test adding labels at the beginning.
|
|
{
|
|
labels: labels.FromStrings("c", "d"),
|
|
externalLabels: []labels.Label{{Name: "a", Value: "b"}},
|
|
expected: labels.FromStrings("a", "b", "c", "d"),
|
|
},
|
|
|
|
// Test we don't override existing labels.
|
|
{
|
|
labels: labels.FromStrings("a", "b"),
|
|
externalLabels: []labels.Label{{Name: "a", Value: "c"}},
|
|
expected: labels.FromStrings("a", "b"),
|
|
},
|
|
|
|
// Test empty externalLabels.
|
|
{
|
|
labels: labels.FromStrings("a", "b"),
|
|
externalLabels: []labels.Label{},
|
|
expected: labels.FromStrings("a", "b"),
|
|
},
|
|
|
|
// Test empty labels.
|
|
{
|
|
labels: labels.EmptyLabels(),
|
|
externalLabels: []labels.Label{{Name: "a", Value: "b"}},
|
|
expected: labels.FromStrings("a", "b"),
|
|
},
|
|
|
|
// Test labels is longer than externalLabels.
|
|
{
|
|
labels: labels.FromStrings("a", "b", "c", "d"),
|
|
externalLabels: []labels.Label{{Name: "e", Value: "f"}},
|
|
expected: labels.FromStrings("a", "b", "c", "d", "e", "f"),
|
|
},
|
|
|
|
// Test externalLabels is longer than labels.
|
|
{
|
|
labels: labels.FromStrings("c", "d"),
|
|
externalLabels: []labels.Label{{Name: "a", Value: "b"}, {Name: "e", Value: "f"}},
|
|
expected: labels.FromStrings("a", "b", "c", "d", "e", "f"),
|
|
},
|
|
|
|
// Adding with and without clashing labels.
|
|
{
|
|
labels: labels.FromStrings("a", "b", "c", "d"),
|
|
externalLabels: []labels.Label{{Name: "a", Value: "xxx"}, {Name: "c", Value: "yyy"}, {Name: "e", Value: "f"}},
|
|
expected: labels.FromStrings("a", "b", "c", "d", "e", "f"),
|
|
},
|
|
} {
|
|
b.Reset(tc.labels)
|
|
processExternalLabels(b, tc.externalLabels)
|
|
testutil.RequireEqual(t, tc.expected, b.Labels(), "test %d", i)
|
|
}
|
|
}
|
|
|
|
func TestCalculateDesiredShards(t *testing.T) {
|
|
cfg := config.DefaultQueueConfig
|
|
_, m := newTestClientAndQueueManager(t, defaultFlushDeadline, config.RemoteWriteProtoMsgV1)
|
|
samplesIn := m.dataIn
|
|
|
|
// Need to start the queue manager so the proper metrics are initialized.
|
|
// However we can stop it right away since we don't need to do any actual
|
|
// processing.
|
|
m.Start()
|
|
m.Stop()
|
|
|
|
inputRate := int64(50000)
|
|
var pendingSamples int64
|
|
|
|
// Two minute startup, no samples are sent.
|
|
startedAt := time.Now().Add(-2 * time.Minute)
|
|
|
|
// helper function for adding samples.
|
|
addSamples := func(s int64, ts time.Duration) {
|
|
pendingSamples += s
|
|
samplesIn.incr(s)
|
|
samplesIn.tick()
|
|
|
|
m.highestRecvTimestamp.Set(float64(startedAt.Add(ts).Unix()))
|
|
}
|
|
|
|
// helper function for sending samples.
|
|
sendSamples := func(s int64, ts time.Duration) {
|
|
pendingSamples -= s
|
|
m.dataOut.incr(s)
|
|
m.dataOutDuration.incr(int64(m.numShards) * int64(shardUpdateDuration))
|
|
|
|
// highest sent is how far back pending samples would be at our input rate.
|
|
highestSent := startedAt.Add(ts - time.Duration(pendingSamples/inputRate)*time.Second)
|
|
m.metrics.highestSentTimestamp.Set(float64(highestSent.Unix()))
|
|
|
|
m.lastSendTimestamp.Store(time.Now().Unix())
|
|
}
|
|
|
|
ts := time.Duration(0)
|
|
for ; ts < 120*time.Second; ts += shardUpdateDuration {
|
|
addSamples(inputRate*int64(shardUpdateDuration/time.Second), ts)
|
|
m.numShards = m.calculateDesiredShards()
|
|
require.Equal(t, 1, m.numShards)
|
|
}
|
|
|
|
// Assume 100ms per request, or 10 requests per second per shard.
|
|
// Shard calculation should never drop below barely keeping up.
|
|
minShards := int(inputRate) / cfg.MaxSamplesPerSend / 10
|
|
// This test should never go above 200 shards, that would be more resources than needed.
|
|
maxShards := 200
|
|
|
|
for ; ts < 15*time.Minute; ts += shardUpdateDuration {
|
|
sin := inputRate * int64(shardUpdateDuration/time.Second)
|
|
addSamples(sin, ts)
|
|
|
|
sout := int64(m.numShards*cfg.MaxSamplesPerSend) * int64(shardUpdateDuration/(100*time.Millisecond))
|
|
// You can't send samples that don't exist so cap at the number of pending samples.
|
|
if sout > pendingSamples {
|
|
sout = pendingSamples
|
|
}
|
|
sendSamples(sout, ts)
|
|
|
|
t.Log("desiredShards", m.numShards, "pendingSamples", pendingSamples)
|
|
m.numShards = m.calculateDesiredShards()
|
|
require.GreaterOrEqual(t, m.numShards, minShards, "Shards are too low. desiredShards=%d, minShards=%d, t_seconds=%d", m.numShards, minShards, ts/time.Second)
|
|
require.LessOrEqual(t, m.numShards, maxShards, "Shards are too high. desiredShards=%d, maxShards=%d, t_seconds=%d", m.numShards, maxShards, ts/time.Second)
|
|
}
|
|
require.Equal(t, int64(0), pendingSamples, "Remote write never caught up, there are still %d pending samples.", pendingSamples)
|
|
}
|
|
|
|
func TestCalculateDesiredShardsDetail(t *testing.T) {
|
|
_, m := newTestClientAndQueueManager(t, defaultFlushDeadline, config.RemoteWriteProtoMsgV1)
|
|
samplesIn := m.dataIn
|
|
|
|
for _, tc := range []struct {
|
|
name string
|
|
prevShards int
|
|
dataIn int64 // Quantities normalised to seconds.
|
|
dataOut int64
|
|
dataDropped int64
|
|
dataOutDuration float64
|
|
backlog float64
|
|
expectedShards int
|
|
}{
|
|
{
|
|
name: "nothing in or out 1",
|
|
prevShards: 1,
|
|
expectedShards: 1, // Shards stays the same.
|
|
},
|
|
{
|
|
name: "nothing in or out 10",
|
|
prevShards: 10,
|
|
expectedShards: 10, // Shards stays the same.
|
|
},
|
|
{
|
|
name: "steady throughput",
|
|
prevShards: 1,
|
|
dataIn: 10,
|
|
dataOut: 10,
|
|
dataOutDuration: 1,
|
|
expectedShards: 1,
|
|
},
|
|
{
|
|
name: "scale down",
|
|
prevShards: 10,
|
|
dataIn: 10,
|
|
dataOut: 10,
|
|
dataOutDuration: 5,
|
|
expectedShards: 5,
|
|
},
|
|
{
|
|
name: "scale down constrained",
|
|
prevShards: 7,
|
|
dataIn: 10,
|
|
dataOut: 10,
|
|
dataOutDuration: 5,
|
|
expectedShards: 7,
|
|
},
|
|
{
|
|
name: "scale up",
|
|
prevShards: 1,
|
|
dataIn: 10,
|
|
dataOut: 10,
|
|
dataOutDuration: 10,
|
|
expectedShards: 10,
|
|
},
|
|
{
|
|
name: "scale up constrained",
|
|
prevShards: 8,
|
|
dataIn: 10,
|
|
dataOut: 10,
|
|
dataOutDuration: 10,
|
|
expectedShards: 8,
|
|
},
|
|
{
|
|
name: "backlogged 20s",
|
|
prevShards: 2,
|
|
dataIn: 10,
|
|
dataOut: 10,
|
|
dataOutDuration: 2,
|
|
backlog: 20,
|
|
expectedShards: 4,
|
|
},
|
|
{
|
|
name: "backlogged 90s",
|
|
prevShards: 4,
|
|
dataIn: 10,
|
|
dataOut: 10,
|
|
dataOutDuration: 4,
|
|
backlog: 90,
|
|
expectedShards: 22,
|
|
},
|
|
{
|
|
name: "backlog reduced",
|
|
prevShards: 22,
|
|
dataIn: 10,
|
|
dataOut: 20,
|
|
dataOutDuration: 4,
|
|
backlog: 10,
|
|
expectedShards: 3,
|
|
},
|
|
{
|
|
name: "backlog eliminated",
|
|
prevShards: 3,
|
|
dataIn: 10,
|
|
dataOut: 10,
|
|
dataOutDuration: 2,
|
|
backlog: 0,
|
|
expectedShards: 2, // Shard back down.
|
|
},
|
|
{
|
|
name: "slight slowdown",
|
|
prevShards: 1,
|
|
dataIn: 10,
|
|
dataOut: 10,
|
|
dataOutDuration: 1.2,
|
|
expectedShards: 2, // 1.2 is rounded up to 2.
|
|
},
|
|
{
|
|
name: "bigger slowdown",
|
|
prevShards: 1,
|
|
dataIn: 10,
|
|
dataOut: 10,
|
|
dataOutDuration: 1.4,
|
|
expectedShards: 2,
|
|
},
|
|
{
|
|
name: "speed up",
|
|
prevShards: 2,
|
|
dataIn: 10,
|
|
dataOut: 10,
|
|
dataOutDuration: 1.2,
|
|
backlog: 0,
|
|
expectedShards: 2, // No reaction - 1.2 is rounded up to 2.
|
|
},
|
|
{
|
|
name: "speed up more",
|
|
prevShards: 2,
|
|
dataIn: 10,
|
|
dataOut: 10,
|
|
dataOutDuration: 0.9,
|
|
backlog: 0,
|
|
expectedShards: 1,
|
|
},
|
|
{
|
|
name: "marginal decision A",
|
|
prevShards: 3,
|
|
dataIn: 10,
|
|
dataOut: 10,
|
|
dataOutDuration: 2.01,
|
|
backlog: 0,
|
|
expectedShards: 3, // 2.01 rounds up to 3.
|
|
},
|
|
{
|
|
name: "marginal decision B",
|
|
prevShards: 3,
|
|
dataIn: 10,
|
|
dataOut: 10,
|
|
dataOutDuration: 1.99,
|
|
backlog: 0,
|
|
expectedShards: 2, // 1.99 rounds up to 2.
|
|
},
|
|
} {
|
|
t.Run(tc.name, func(t *testing.T) {
|
|
m.numShards = tc.prevShards
|
|
forceEMWA(samplesIn, tc.dataIn*int64(shardUpdateDuration/time.Second))
|
|
samplesIn.tick()
|
|
forceEMWA(m.dataOut, tc.dataOut*int64(shardUpdateDuration/time.Second))
|
|
forceEMWA(m.dataDropped, tc.dataDropped*int64(shardUpdateDuration/time.Second))
|
|
forceEMWA(m.dataOutDuration, int64(tc.dataOutDuration*float64(shardUpdateDuration)))
|
|
m.highestRecvTimestamp.value = tc.backlog // Not Set() because it can only increase value.
|
|
|
|
require.Equal(t, tc.expectedShards, m.calculateDesiredShards())
|
|
})
|
|
}
|
|
}
|
|
|
|
func forceEMWA(r *ewmaRate, rate int64) {
|
|
r.init = false
|
|
r.newEvents.Store(rate)
|
|
}
|
|
|
|
func TestQueueManagerMetrics(t *testing.T) {
|
|
reg := prometheus.NewPedanticRegistry()
|
|
metrics := newQueueManagerMetrics(reg, "name", "http://localhost:1234")
|
|
|
|
// Make sure metrics pass linting.
|
|
problems, err := client_testutil.GatherAndLint(reg)
|
|
require.NoError(t, err)
|
|
require.Empty(t, problems, "Metric linting problems detected: %v", problems)
|
|
|
|
// Make sure all metrics were unregistered. A failure here means you need
|
|
// unregister a metric in `queueManagerMetrics.unregister()`.
|
|
metrics.unregister()
|
|
err = client_testutil.GatherAndCompare(reg, strings.NewReader(""))
|
|
require.NoError(t, err)
|
|
}
|
|
|
|
func TestQueue_FlushAndShutdownDoesNotDeadlock(t *testing.T) {
|
|
capacity := 100
|
|
batchSize := 10
|
|
queue := newQueue(batchSize, capacity)
|
|
for i := 0; i < capacity+batchSize; i++ {
|
|
queue.Append(timeSeries{})
|
|
}
|
|
|
|
done := make(chan struct{})
|
|
go queue.FlushAndShutdown(done)
|
|
go func() {
|
|
// Give enough time for FlushAndShutdown to acquire the lock. queue.Batch()
|
|
// should not block forever even if the lock is acquired.
|
|
time.Sleep(10 * time.Millisecond)
|
|
queue.Batch()
|
|
close(done)
|
|
}()
|
|
select {
|
|
case <-done:
|
|
case <-time.After(2 * time.Second):
|
|
t.Error("Deadlock in FlushAndShutdown detected")
|
|
pprof.Lookup("goroutine").WriteTo(os.Stdout, 1)
|
|
t.FailNow()
|
|
}
|
|
}
|
|
|
|
func createDummyTimeSeries(instances int) []timeSeries {
|
|
metrics := []labels.Labels{
|
|
labels.FromStrings("__name__", "go_gc_duration_seconds", "quantile", "0"),
|
|
labels.FromStrings("__name__", "go_gc_duration_seconds", "quantile", "0.25"),
|
|
labels.FromStrings("__name__", "go_gc_duration_seconds", "quantile", "0.5"),
|
|
labels.FromStrings("__name__", "go_gc_duration_seconds", "quantile", "0.75"),
|
|
labels.FromStrings("__name__", "go_gc_duration_seconds", "quantile", "1"),
|
|
labels.FromStrings("__name__", "go_gc_duration_seconds_sum"),
|
|
labels.FromStrings("__name__", "go_gc_duration_seconds_count"),
|
|
labels.FromStrings("__name__", "go_memstats_alloc_bytes_total"),
|
|
labels.FromStrings("__name__", "go_memstats_frees_total"),
|
|
labels.FromStrings("__name__", "go_memstats_lookups_total"),
|
|
labels.FromStrings("__name__", "go_memstats_mallocs_total"),
|
|
labels.FromStrings("__name__", "go_goroutines"),
|
|
labels.FromStrings("__name__", "go_info", "version", "go1.19.3"),
|
|
labels.FromStrings("__name__", "go_memstats_alloc_bytes"),
|
|
labels.FromStrings("__name__", "go_memstats_buck_hash_sys_bytes"),
|
|
labels.FromStrings("__name__", "go_memstats_gc_sys_bytes"),
|
|
labels.FromStrings("__name__", "go_memstats_heap_alloc_bytes"),
|
|
labels.FromStrings("__name__", "go_memstats_heap_idle_bytes"),
|
|
labels.FromStrings("__name__", "go_memstats_heap_inuse_bytes"),
|
|
labels.FromStrings("__name__", "go_memstats_heap_objects"),
|
|
labels.FromStrings("__name__", "go_memstats_heap_released_bytes"),
|
|
labels.FromStrings("__name__", "go_memstats_heap_sys_bytes"),
|
|
labels.FromStrings("__name__", "go_memstats_last_gc_time_seconds"),
|
|
labels.FromStrings("__name__", "go_memstats_mcache_inuse_bytes"),
|
|
labels.FromStrings("__name__", "go_memstats_mcache_sys_bytes"),
|
|
labels.FromStrings("__name__", "go_memstats_mspan_inuse_bytes"),
|
|
labels.FromStrings("__name__", "go_memstats_mspan_sys_bytes"),
|
|
labels.FromStrings("__name__", "go_memstats_next_gc_bytes"),
|
|
labels.FromStrings("__name__", "go_memstats_other_sys_bytes"),
|
|
labels.FromStrings("__name__", "go_memstats_stack_inuse_bytes"),
|
|
labels.FromStrings("__name__", "go_memstats_stack_sys_bytes"),
|
|
labels.FromStrings("__name__", "go_memstats_sys_bytes"),
|
|
labels.FromStrings("__name__", "go_threads"),
|
|
}
|
|
|
|
commonLabels := labels.FromStrings(
|
|
"cluster", "some-cluster-0",
|
|
"container", "prometheus",
|
|
"job", "some-namespace/prometheus",
|
|
"namespace", "some-namespace")
|
|
|
|
var result []timeSeries
|
|
r := rand.New(rand.NewSource(0))
|
|
for i := 0; i < instances; i++ {
|
|
b := labels.NewBuilder(commonLabels)
|
|
b.Set("pod", "prometheus-"+strconv.Itoa(i))
|
|
for _, lbls := range metrics {
|
|
lbls.Range(func(l labels.Label) {
|
|
b.Set(l.Name, l.Value)
|
|
})
|
|
result = append(result, timeSeries{
|
|
seriesLabels: b.Labels(),
|
|
value: r.Float64(),
|
|
})
|
|
}
|
|
}
|
|
return result
|
|
}
|
|
|
|
func BenchmarkBuildWriteRequest(b *testing.B) {
|
|
noopLogger := log.NewNopLogger()
|
|
bench := func(b *testing.B, batch []timeSeries) {
|
|
buff := make([]byte, 0)
|
|
seriesBuff := make([]prompb.TimeSeries, len(batch))
|
|
for i := range seriesBuff {
|
|
seriesBuff[i].Samples = []prompb.Sample{{}}
|
|
seriesBuff[i].Exemplars = []prompb.Exemplar{{}}
|
|
}
|
|
pBuf := proto.NewBuffer(nil)
|
|
|
|
totalSize := 0
|
|
for i := 0; i < b.N; i++ {
|
|
populateTimeSeries(batch, seriesBuff, true, true)
|
|
req, _, _, err := buildWriteRequest(noopLogger, seriesBuff, nil, pBuf, &buff, nil, "snappy")
|
|
if err != nil {
|
|
b.Fatal(err)
|
|
}
|
|
totalSize += len(req)
|
|
b.ReportMetric(float64(totalSize)/float64(b.N), "compressedSize/op")
|
|
}
|
|
}
|
|
|
|
twoBatch := createDummyTimeSeries(2)
|
|
tenBatch := createDummyTimeSeries(10)
|
|
hundredBatch := createDummyTimeSeries(100)
|
|
|
|
b.Run("2 instances", func(b *testing.B) {
|
|
bench(b, twoBatch)
|
|
})
|
|
|
|
b.Run("10 instances", func(b *testing.B) {
|
|
bench(b, tenBatch)
|
|
})
|
|
|
|
b.Run("1k instances", func(b *testing.B) {
|
|
bench(b, hundredBatch)
|
|
})
|
|
}
|
|
|
|
func BenchmarkBuildV2WriteRequest(b *testing.B) {
|
|
noopLogger := log.NewNopLogger()
|
|
bench := func(b *testing.B, batch []timeSeries) {
|
|
symbolTable := writev2.NewSymbolTable()
|
|
buff := make([]byte, 0)
|
|
seriesBuff := make([]writev2.TimeSeries, len(batch))
|
|
for i := range seriesBuff {
|
|
seriesBuff[i].Samples = []writev2.Sample{{}}
|
|
seriesBuff[i].Exemplars = []writev2.Exemplar{{}}
|
|
}
|
|
pBuf := []byte{}
|
|
|
|
totalSize := 0
|
|
for i := 0; i < b.N; i++ {
|
|
populateV2TimeSeries(&symbolTable, batch, seriesBuff, true, true)
|
|
req, _, _, err := buildV2WriteRequest(noopLogger, seriesBuff, symbolTable.Symbols(), &pBuf, &buff, nil, "snappy")
|
|
if err != nil {
|
|
b.Fatal(err)
|
|
}
|
|
totalSize += len(req)
|
|
b.ReportMetric(float64(totalSize)/float64(b.N), "compressedSize/op")
|
|
}
|
|
}
|
|
|
|
twoBatch := createDummyTimeSeries(2)
|
|
tenBatch := createDummyTimeSeries(10)
|
|
hundredBatch := createDummyTimeSeries(100)
|
|
|
|
b.Run("2 instances", func(b *testing.B) {
|
|
bench(b, twoBatch)
|
|
})
|
|
|
|
b.Run("10 instances", func(b *testing.B) {
|
|
bench(b, tenBatch)
|
|
})
|
|
|
|
b.Run("1k instances", func(b *testing.B) {
|
|
bench(b, hundredBatch)
|
|
})
|
|
}
|
|
|
|
func TestDropOldTimeSeries(t *testing.T) {
|
|
size := 10
|
|
nSeries := 6
|
|
nSamples := config.DefaultQueueConfig.Capacity * size
|
|
samples, newSamples, series := createTimeseriesWithOldSamples(nSamples, nSeries)
|
|
|
|
// TODO(alexg): test with new version
|
|
c := NewTestWriteClient(config.RemoteWriteProtoMsgV1)
|
|
c.expectSamples(newSamples, series)
|
|
|
|
cfg := config.DefaultQueueConfig
|
|
mcfg := config.DefaultMetadataConfig
|
|
cfg.MaxShards = 1
|
|
cfg.SampleAgeLimit = model.Duration(60 * time.Second)
|
|
m := newTestQueueManager(t, cfg, mcfg, defaultFlushDeadline, c, config.RemoteWriteProtoMsgV1)
|
|
m.StoreSeries(series, 0)
|
|
|
|
m.Start()
|
|
defer m.Stop()
|
|
|
|
m.Append(samples)
|
|
c.waitForExpectedData(t, 30*time.Second)
|
|
}
|
|
|
|
func TestIsSampleOld(t *testing.T) {
|
|
currentTime := time.Now()
|
|
require.True(t, isSampleOld(currentTime, 60*time.Second, timestamp.FromTime(currentTime.Add(-61*time.Second))))
|
|
require.False(t, isSampleOld(currentTime, 60*time.Second, timestamp.FromTime(currentTime.Add(-59*time.Second))))
|
|
}
|
|
|
|
// Simulates scenario in which remote write endpoint is down and a subset of samples is dropped due to age limit while backoffing.
|
|
func TestSendSamplesWithBackoffWithSampleAgeLimit(t *testing.T) {
|
|
maxSamplesPerSend := 10
|
|
sampleAgeLimit := time.Second
|
|
|
|
cfg := config.DefaultQueueConfig
|
|
cfg.MaxShards = 1
|
|
cfg.SampleAgeLimit = model.Duration(sampleAgeLimit)
|
|
// Set the batch send deadline to 5 minutes to effectively disable it.
|
|
cfg.BatchSendDeadline = model.Duration(time.Minute * 5)
|
|
cfg.Capacity = 10 * maxSamplesPerSend // more than the amount of data we append in the test
|
|
cfg.MaxBackoff = model.Duration(time.Millisecond * 100)
|
|
cfg.MinBackoff = model.Duration(time.Millisecond * 100)
|
|
cfg.MaxSamplesPerSend = maxSamplesPerSend
|
|
metadataCfg := config.DefaultMetadataConfig
|
|
metadataCfg.Send = true
|
|
metadataCfg.SendInterval = model.Duration(time.Second * 60)
|
|
metadataCfg.MaxSamplesPerSend = maxSamplesPerSend
|
|
c := NewTestWriteClient(config.RemoteWriteProtoMsgV1)
|
|
m := newTestQueueManager(t, cfg, metadataCfg, time.Second, c, config.RemoteWriteProtoMsgV1)
|
|
|
|
m.Start()
|
|
|
|
batchID := 0
|
|
expectedSamples := map[string][]prompb.Sample{}
|
|
|
|
appendData := func(numberOfSeries int, timeAdd time.Duration, shouldBeDropped bool) {
|
|
t.Log(">>>> Appending series ", numberOfSeries, " as batch ID ", batchID, " with timeAdd ", timeAdd, " and should be dropped ", shouldBeDropped)
|
|
samples, series := createTimeseriesWithRandomLabelCount(strconv.Itoa(batchID), numberOfSeries, timeAdd, 9)
|
|
m.StoreSeries(series, batchID)
|
|
sent := m.Append(samples)
|
|
require.True(t, sent, "samples not sent")
|
|
if !shouldBeDropped {
|
|
for _, s := range samples {
|
|
tsID := getSeriesIDFromRef(series[s.Ref])
|
|
expectedSamples[tsID] = append(c.expectedSamples[tsID], prompb.Sample{
|
|
Timestamp: s.T,
|
|
Value: s.V,
|
|
})
|
|
}
|
|
}
|
|
batchID++
|
|
}
|
|
timeShift := -time.Millisecond * 5
|
|
|
|
c.SetReturnError(RecoverableError{context.DeadlineExceeded, defaultBackoff})
|
|
|
|
appendData(maxSamplesPerSend/2, timeShift, true)
|
|
time.Sleep(sampleAgeLimit)
|
|
appendData(maxSamplesPerSend/2, timeShift, true)
|
|
time.Sleep(sampleAgeLimit / 10)
|
|
appendData(maxSamplesPerSend/2, timeShift, true)
|
|
time.Sleep(2 * sampleAgeLimit)
|
|
appendData(2*maxSamplesPerSend, timeShift, false)
|
|
time.Sleep(sampleAgeLimit / 2)
|
|
c.SetReturnError(nil)
|
|
appendData(5, timeShift, false)
|
|
m.Stop()
|
|
|
|
if diff := cmp.Diff(expectedSamples, c.receivedSamples); diff != "" {
|
|
t.Errorf("mismatch (-want +got):\n%s", diff)
|
|
}
|
|
}
|
|
|
|
func createTimeseriesWithRandomLabelCount(id string, seriesCount int, timeAdd time.Duration, maxLabels int) ([]record.RefSample, []record.RefSeries) {
|
|
samples := []record.RefSample{}
|
|
series := []record.RefSeries{}
|
|
// use a fixed rand source so tests are consistent
|
|
r := rand.New(rand.NewSource(99))
|
|
for i := 0; i < seriesCount; i++ {
|
|
s := record.RefSample{
|
|
Ref: chunks.HeadSeriesRef(i),
|
|
T: time.Now().Add(timeAdd).UnixMilli(),
|
|
V: r.Float64(),
|
|
}
|
|
samples = append(samples, s)
|
|
labelsCount := r.Intn(maxLabels)
|
|
lb := labels.NewScratchBuilder(1 + labelsCount)
|
|
lb.Add("__name__", "batch_"+id+"_id_"+strconv.Itoa(i))
|
|
for j := 1; j < labelsCount+1; j++ {
|
|
// same for both name and value
|
|
label := "batch_" + id + "_label_" + strconv.Itoa(j)
|
|
lb.Add(label, label)
|
|
}
|
|
series = append(series, record.RefSeries{
|
|
Ref: chunks.HeadSeriesRef(i),
|
|
Labels: lb.Labels(),
|
|
})
|
|
}
|
|
return samples, series
|
|
}
|
|
|
|
func createTimeseriesWithOldSamples(numSamples, numSeries int, extraLabels ...labels.Label) ([]record.RefSample, []record.RefSample, []record.RefSeries) {
|
|
newSamples := make([]record.RefSample, 0, numSamples)
|
|
samples := make([]record.RefSample, 0, numSamples)
|
|
series := make([]record.RefSeries, 0, numSeries)
|
|
lb := labels.NewScratchBuilder(1 + len(extraLabels))
|
|
for i := 0; i < numSeries; i++ {
|
|
name := fmt.Sprintf("test_metric_%d", i)
|
|
// We create half of the samples in the past.
|
|
past := timestamp.FromTime(time.Now().Add(-5 * time.Minute))
|
|
for j := 0; j < numSamples/2; j++ {
|
|
samples = append(samples, record.RefSample{
|
|
Ref: chunks.HeadSeriesRef(i),
|
|
T: past + int64(j),
|
|
V: float64(i),
|
|
})
|
|
}
|
|
for j := 0; j < numSamples/2; j++ {
|
|
sample := record.RefSample{
|
|
Ref: chunks.HeadSeriesRef(i),
|
|
T: int64(int(time.Now().UnixMilli()) + j),
|
|
V: float64(i),
|
|
}
|
|
samples = append(samples, sample)
|
|
newSamples = append(newSamples, sample)
|
|
}
|
|
// Create Labels that is name of series plus any extra labels supplied.
|
|
lb.Reset()
|
|
lb.Add(labels.MetricName, name)
|
|
for _, l := range extraLabels {
|
|
lb.Add(l.Name, l.Value)
|
|
}
|
|
lb.Sort()
|
|
series = append(series, record.RefSeries{
|
|
Ref: chunks.HeadSeriesRef(i),
|
|
Labels: lb.Labels(),
|
|
})
|
|
}
|
|
return samples, newSamples, series
|
|
}
|
|
|
|
func filterTsLimit(limit int64, ts prompb.TimeSeries) bool {
|
|
return limit > ts.Samples[0].Timestamp
|
|
}
|
|
|
|
func TestBuildTimeSeries(t *testing.T) {
|
|
testCases := []struct {
|
|
name string
|
|
ts []prompb.TimeSeries
|
|
filter func(ts prompb.TimeSeries) bool
|
|
lowestTs int64
|
|
highestTs int64
|
|
droppedSamples int
|
|
responseLen int
|
|
}{
|
|
{
|
|
name: "No filter applied",
|
|
ts: []prompb.TimeSeries{
|
|
{
|
|
Samples: []prompb.Sample{
|
|
{
|
|
Timestamp: 1234567890,
|
|
Value: 1.23,
|
|
},
|
|
},
|
|
},
|
|
{
|
|
Samples: []prompb.Sample{
|
|
{
|
|
Timestamp: 1234567891,
|
|
Value: 2.34,
|
|
},
|
|
},
|
|
},
|
|
{
|
|
Samples: []prompb.Sample{
|
|
{
|
|
Timestamp: 1234567892,
|
|
Value: 3.34,
|
|
},
|
|
},
|
|
},
|
|
},
|
|
filter: nil,
|
|
responseLen: 3,
|
|
lowestTs: 1234567890,
|
|
highestTs: 1234567892,
|
|
},
|
|
{
|
|
name: "Filter applied, samples in order",
|
|
ts: []prompb.TimeSeries{
|
|
{
|
|
Samples: []prompb.Sample{
|
|
{
|
|
Timestamp: 1234567890,
|
|
Value: 1.23,
|
|
},
|
|
},
|
|
},
|
|
{
|
|
Samples: []prompb.Sample{
|
|
{
|
|
Timestamp: 1234567891,
|
|
Value: 2.34,
|
|
},
|
|
},
|
|
},
|
|
{
|
|
Samples: []prompb.Sample{
|
|
{
|
|
Timestamp: 1234567892,
|
|
Value: 3.45,
|
|
},
|
|
},
|
|
},
|
|
{
|
|
Samples: []prompb.Sample{
|
|
{
|
|
Timestamp: 1234567893,
|
|
Value: 3.45,
|
|
},
|
|
},
|
|
},
|
|
},
|
|
filter: func(ts prompb.TimeSeries) bool { return filterTsLimit(1234567892, ts) },
|
|
responseLen: 2,
|
|
lowestTs: 1234567892,
|
|
highestTs: 1234567893,
|
|
droppedSamples: 2,
|
|
},
|
|
{
|
|
name: "Filter applied, samples out of order",
|
|
ts: []prompb.TimeSeries{
|
|
{
|
|
Samples: []prompb.Sample{
|
|
{
|
|
Timestamp: 1234567892,
|
|
Value: 3.45,
|
|
},
|
|
},
|
|
},
|
|
{
|
|
Samples: []prompb.Sample{
|
|
{
|
|
Timestamp: 1234567890,
|
|
Value: 1.23,
|
|
},
|
|
},
|
|
},
|
|
{
|
|
Samples: []prompb.Sample{
|
|
{
|
|
Timestamp: 1234567893,
|
|
Value: 3.45,
|
|
},
|
|
},
|
|
},
|
|
{
|
|
Samples: []prompb.Sample{
|
|
{
|
|
Timestamp: 1234567891,
|
|
Value: 2.34,
|
|
},
|
|
},
|
|
},
|
|
},
|
|
filter: func(ts prompb.TimeSeries) bool { return filterTsLimit(1234567892, ts) },
|
|
responseLen: 2,
|
|
lowestTs: 1234567892,
|
|
highestTs: 1234567893,
|
|
droppedSamples: 2,
|
|
},
|
|
{
|
|
name: "Filter applied, samples not consecutive",
|
|
ts: []prompb.TimeSeries{
|
|
{
|
|
Samples: []prompb.Sample{
|
|
{
|
|
Timestamp: 1234567890,
|
|
Value: 1.23,
|
|
},
|
|
},
|
|
},
|
|
{
|
|
Samples: []prompb.Sample{
|
|
{
|
|
Timestamp: 1234567892,
|
|
Value: 3.45,
|
|
},
|
|
},
|
|
},
|
|
{
|
|
Samples: []prompb.Sample{
|
|
{
|
|
Timestamp: 1234567895,
|
|
Value: 6.78,
|
|
},
|
|
},
|
|
},
|
|
{
|
|
Samples: []prompb.Sample{
|
|
{
|
|
Timestamp: 1234567897,
|
|
Value: 6.78,
|
|
},
|
|
},
|
|
},
|
|
},
|
|
filter: func(ts prompb.TimeSeries) bool { return filterTsLimit(1234567895, ts) },
|
|
responseLen: 2,
|
|
lowestTs: 1234567895,
|
|
highestTs: 1234567897,
|
|
droppedSamples: 2,
|
|
},
|
|
}
|
|
|
|
// Run the test cases
|
|
for _, tc := range testCases {
|
|
t.Run(tc.name, func(t *testing.T) {
|
|
highest, lowest, result, droppedSamples, _, _ := buildTimeSeries(tc.ts, tc.filter)
|
|
require.NotNil(t, result)
|
|
require.Len(t, result, tc.responseLen)
|
|
require.Equal(t, tc.highestTs, highest)
|
|
require.Equal(t, tc.lowestTs, lowest)
|
|
require.Equal(t, tc.droppedSamples, droppedSamples)
|
|
})
|
|
}
|
|
}
|
|
|
|
func BenchmarkBuildTimeSeries(b *testing.B) {
|
|
// Send one sample per series, which is the typical remote_write case
|
|
const numSamples = 10000
|
|
filter := func(ts prompb.TimeSeries) bool { return filterTsLimit(99, ts) }
|
|
for i := 0; i < b.N; i++ {
|
|
samples := createProtoTimeseriesWithOld(numSamples, 100, extraLabels...)
|
|
_, _, result, _, _, _ := buildTimeSeries(samples, filter)
|
|
require.NotNil(b, result)
|
|
}
|
|
}
|