2019-05-22 23:16:55 +02:00
|
|
|
package storage
|
|
|
|
|
|
|
|
import (
|
|
|
|
"fmt"
|
|
|
|
"math/rand"
|
|
|
|
"os"
|
|
|
|
"reflect"
|
2020-11-15 23:42:27 +01:00
|
|
|
"sort"
|
2019-05-22 23:16:55 +02:00
|
|
|
"strings"
|
|
|
|
"testing"
|
|
|
|
"testing/quick"
|
|
|
|
"time"
|
2019-09-24 20:10:22 +02:00
|
|
|
|
|
|
|
"github.com/VictoriaMetrics/VictoriaMetrics/lib/uint64set"
|
2019-05-22 23:16:55 +02:00
|
|
|
)
|
|
|
|
|
2021-02-02 23:24:05 +01:00
|
|
|
func TestGetRegexpForGraphiteNodeQuery(t *testing.T) {
|
|
|
|
f := func(q, expectedRegexp string) {
|
|
|
|
t.Helper()
|
2021-02-03 19:12:17 +01:00
|
|
|
re, err := getRegexpForGraphiteQuery(q)
|
2021-02-02 23:24:05 +01:00
|
|
|
if err != nil {
|
|
|
|
t.Fatalf("unexpected error for query=%q: %s", q, err)
|
|
|
|
}
|
|
|
|
reStr := re.String()
|
|
|
|
if reStr != expectedRegexp {
|
|
|
|
t.Fatalf("unexpected regexp for query %q; got %q want %q", q, reStr, expectedRegexp)
|
|
|
|
}
|
|
|
|
}
|
|
|
|
f(``, `^$`)
|
|
|
|
f(`*`, `^[^.]*$`)
|
|
|
|
f(`foo.`, `^foo\.$`)
|
|
|
|
f(`foo.bar`, `^foo\.bar$`)
|
|
|
|
f(`{foo,b*ar,b[a-z]}`, `^(?:foo|b[^.]*ar|b[a-z])$`)
|
|
|
|
f(`[-a-zx.]`, `^[-a-zx.]$`)
|
|
|
|
f(`**`, `^[^.]*[^.]*$`)
|
|
|
|
f(`a*[de]{x,y}z`, `^a[^.]*[de](?:x|y)z$`)
|
2021-02-03 19:12:17 +01:00
|
|
|
f(`foo{bar`, `^foo\{bar$`)
|
|
|
|
f(`foo{ba,r`, `^foo\{ba,r$`)
|
|
|
|
f(`foo[bar`, `^foo\[bar$`)
|
|
|
|
f(`foo{bar}`, `^foobar$`)
|
|
|
|
f(`foo{bar,,b{{a,b*},z},[x-y]*z}a`, `^foo(?:bar||b(?:(?:a|b[^.]*)|z)|[x-y][^.]*z)a$`)
|
2021-02-02 23:24:05 +01:00
|
|
|
}
|
|
|
|
|
2019-11-11 12:21:05 +01:00
|
|
|
func TestDateMetricIDCacheSerial(t *testing.T) {
|
|
|
|
c := newDateMetricIDCache()
|
|
|
|
if err := testDateMetricIDCache(c, false); err != nil {
|
|
|
|
t.Fatalf("unexpected error: %s", err)
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
func TestDateMetricIDCacheConcurrent(t *testing.T) {
|
|
|
|
c := newDateMetricIDCache()
|
|
|
|
ch := make(chan error, 5)
|
|
|
|
for i := 0; i < 5; i++ {
|
|
|
|
go func() {
|
|
|
|
ch <- testDateMetricIDCache(c, true)
|
|
|
|
}()
|
|
|
|
}
|
|
|
|
for i := 0; i < 5; i++ {
|
|
|
|
select {
|
|
|
|
case err := <-ch:
|
|
|
|
if err != nil {
|
|
|
|
t.Fatalf("unexpected error: %s", err)
|
|
|
|
}
|
|
|
|
case <-time.After(time.Second * 5):
|
|
|
|
t.Fatalf("timeout")
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
func testDateMetricIDCache(c *dateMetricIDCache, concurrent bool) error {
|
|
|
|
type dmk struct {
|
|
|
|
date uint64
|
|
|
|
metricID uint64
|
|
|
|
}
|
|
|
|
m := make(map[dmk]bool)
|
|
|
|
for i := 0; i < 1e5; i++ {
|
|
|
|
date := uint64(i) % 3
|
|
|
|
metricID := uint64(i) % 1237
|
|
|
|
if !concurrent && c.Has(date, metricID) {
|
|
|
|
if !m[dmk{date, metricID}] {
|
|
|
|
return fmt.Errorf("c.Has(%d, %d) must return false, but returned true", date, metricID)
|
|
|
|
}
|
|
|
|
continue
|
|
|
|
}
|
|
|
|
c.Set(date, metricID)
|
|
|
|
m[dmk{date, metricID}] = true
|
|
|
|
if !concurrent && !c.Has(date, metricID) {
|
|
|
|
return fmt.Errorf("c.Has(%d, %d) must return true, but returned false", date, metricID)
|
|
|
|
}
|
|
|
|
if i%11234 == 0 {
|
|
|
|
c.sync()
|
|
|
|
}
|
|
|
|
if i%34323 == 0 {
|
|
|
|
c.Reset()
|
|
|
|
m = make(map[dmk]bool)
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
// Verify fast path after sync.
|
|
|
|
for i := 0; i < 1e5; i++ {
|
|
|
|
date := uint64(i) % 3
|
|
|
|
metricID := uint64(i) % 123
|
|
|
|
c.Set(date, metricID)
|
|
|
|
}
|
|
|
|
c.sync()
|
|
|
|
for i := 0; i < 1e5; i++ {
|
|
|
|
date := uint64(i) % 3
|
|
|
|
metricID := uint64(i) % 123
|
|
|
|
if !concurrent && !c.Has(date, metricID) {
|
|
|
|
return fmt.Errorf("c.Has(%d, %d) must return true after sync", date, metricID)
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
// Verify c.Reset
|
|
|
|
if n := c.EntriesCount(); !concurrent && n < 123 {
|
|
|
|
return fmt.Errorf("c.EntriesCount must return at least 123; returned %d", n)
|
|
|
|
}
|
|
|
|
c.Reset()
|
|
|
|
if n := c.EntriesCount(); !concurrent && n > 0 {
|
|
|
|
return fmt.Errorf("c.EntriesCount must return 0 after reset; returned %d", n)
|
|
|
|
}
|
|
|
|
return nil
|
|
|
|
}
|
|
|
|
|
2019-06-09 18:06:53 +02:00
|
|
|
func TestUpdateCurrHourMetricIDs(t *testing.T) {
|
|
|
|
newStorage := func() *Storage {
|
|
|
|
var s Storage
|
|
|
|
s.currHourMetricIDs.Store(&hourMetricIDs{})
|
|
|
|
s.prevHourMetricIDs.Store(&hourMetricIDs{})
|
|
|
|
return &s
|
|
|
|
}
|
2019-10-31 14:50:58 +01:00
|
|
|
t.Run("empty_pending_metric_ids_stale_curr_hour", func(t *testing.T) {
|
2019-06-09 18:06:53 +02:00
|
|
|
s := newStorage()
|
|
|
|
hour := uint64(timestampFromTime(time.Now())) / msecPerHour
|
|
|
|
hmOrig := &hourMetricIDs{
|
2019-09-24 20:10:22 +02:00
|
|
|
m: &uint64set.Set{},
|
2019-06-09 18:06:53 +02:00
|
|
|
hour: 123,
|
|
|
|
}
|
2019-09-24 20:10:22 +02:00
|
|
|
hmOrig.m.Add(12)
|
|
|
|
hmOrig.m.Add(34)
|
2019-06-09 18:06:53 +02:00
|
|
|
s.currHourMetricIDs.Store(hmOrig)
|
|
|
|
s.updateCurrHourMetricIDs()
|
|
|
|
hmCurr := s.currHourMetricIDs.Load().(*hourMetricIDs)
|
|
|
|
if hmCurr.hour != hour {
|
2019-11-11 23:16:42 +01:00
|
|
|
// It is possible new hour occurred. Update the hour and verify it again.
|
2019-06-09 18:06:53 +02:00
|
|
|
hour = uint64(timestampFromTime(time.Now())) / msecPerHour
|
|
|
|
if hmCurr.hour != hour {
|
|
|
|
t.Fatalf("unexpected hmCurr.hour; got %d; want %d", hmCurr.hour, hour)
|
|
|
|
}
|
|
|
|
}
|
2019-09-24 20:10:22 +02:00
|
|
|
if hmCurr.m.Len() != 0 {
|
|
|
|
t.Fatalf("unexpected length of hm.m; got %d; want %d", hmCurr.m.Len(), 0)
|
2019-06-09 18:06:53 +02:00
|
|
|
}
|
|
|
|
if !hmCurr.isFull {
|
|
|
|
t.Fatalf("unexpected hmCurr.isFull; got %v; want %v", hmCurr.isFull, true)
|
|
|
|
}
|
|
|
|
|
|
|
|
hmPrev := s.prevHourMetricIDs.Load().(*hourMetricIDs)
|
|
|
|
if !reflect.DeepEqual(hmPrev, hmOrig) {
|
|
|
|
t.Fatalf("unexpected hmPrev; got %v; want %v", hmPrev, hmOrig)
|
|
|
|
}
|
|
|
|
|
2019-11-08 12:16:40 +01:00
|
|
|
if len(s.pendingHourEntries) != 0 {
|
|
|
|
t.Fatalf("unexpected len(s.pendingHourEntries); got %d; want %d", len(s.pendingHourEntries), 0)
|
2019-06-09 18:06:53 +02:00
|
|
|
}
|
|
|
|
})
|
2019-10-31 14:50:58 +01:00
|
|
|
t.Run("empty_pending_metric_ids_valid_curr_hour", func(t *testing.T) {
|
2019-06-09 18:06:53 +02:00
|
|
|
s := newStorage()
|
|
|
|
hour := uint64(timestampFromTime(time.Now())) / msecPerHour
|
|
|
|
hmOrig := &hourMetricIDs{
|
2019-09-24 20:10:22 +02:00
|
|
|
m: &uint64set.Set{},
|
2019-06-09 18:06:53 +02:00
|
|
|
hour: hour,
|
|
|
|
}
|
2019-09-24 20:10:22 +02:00
|
|
|
hmOrig.m.Add(12)
|
|
|
|
hmOrig.m.Add(34)
|
2019-06-09 18:06:53 +02:00
|
|
|
s.currHourMetricIDs.Store(hmOrig)
|
|
|
|
s.updateCurrHourMetricIDs()
|
|
|
|
hmCurr := s.currHourMetricIDs.Load().(*hourMetricIDs)
|
|
|
|
if hmCurr.hour != hour {
|
2019-11-11 23:16:42 +01:00
|
|
|
// It is possible new hour occurred. Update the hour and verify it again.
|
2019-06-09 18:06:53 +02:00
|
|
|
hour = uint64(timestampFromTime(time.Now())) / msecPerHour
|
|
|
|
if hmCurr.hour != hour {
|
|
|
|
t.Fatalf("unexpected hmCurr.hour; got %d; want %d", hmCurr.hour, hour)
|
|
|
|
}
|
|
|
|
// Do not run other checks, since they may fail.
|
|
|
|
return
|
|
|
|
}
|
|
|
|
if !reflect.DeepEqual(hmCurr, hmOrig) {
|
|
|
|
t.Fatalf("unexpected hmCurr; got %v; want %v", hmCurr, hmOrig)
|
|
|
|
}
|
|
|
|
if hmCurr.isFull {
|
|
|
|
t.Fatalf("unexpected hmCurr.isFull; got %v; want %v", hmCurr.isFull, false)
|
|
|
|
}
|
|
|
|
|
|
|
|
hmPrev := s.prevHourMetricIDs.Load().(*hourMetricIDs)
|
|
|
|
hmEmpty := &hourMetricIDs{}
|
|
|
|
if !reflect.DeepEqual(hmPrev, hmEmpty) {
|
|
|
|
t.Fatalf("unexpected hmPrev; got %v; want %v", hmPrev, hmEmpty)
|
|
|
|
}
|
2019-11-08 12:16:40 +01:00
|
|
|
if len(s.pendingHourEntries) != 0 {
|
|
|
|
t.Fatalf("unexpected len(s.pendingHourEntries); got %d; want %d", len(s.pendingHourEntries), 0)
|
2019-06-09 18:06:53 +02:00
|
|
|
}
|
|
|
|
})
|
|
|
|
t.Run("nonempty_pending_metric_ids_stale_curr_hour", func(t *testing.T) {
|
|
|
|
s := newStorage()
|
2019-11-08 12:16:40 +01:00
|
|
|
s.pendingHourEntries = []pendingHourMetricIDEntry{
|
2019-10-31 14:50:58 +01:00
|
|
|
{AccountID: 123, ProjectID: 431, MetricID: 343},
|
|
|
|
{AccountID: 123, ProjectID: 431, MetricID: 32424},
|
|
|
|
{AccountID: 1, ProjectID: 2, MetricID: 8293432},
|
|
|
|
}
|
|
|
|
mExpected := &uint64set.Set{}
|
2019-11-08 12:16:40 +01:00
|
|
|
for _, e := range s.pendingHourEntries {
|
2019-10-31 14:50:58 +01:00
|
|
|
mExpected.Add(e.MetricID)
|
|
|
|
}
|
|
|
|
byTenantExpected := make(map[accountProjectKey]*uint64set.Set)
|
2019-11-08 12:16:40 +01:00
|
|
|
for _, e := range s.pendingHourEntries {
|
2019-10-31 14:50:58 +01:00
|
|
|
k := accountProjectKey{
|
|
|
|
AccountID: e.AccountID,
|
|
|
|
ProjectID: e.ProjectID,
|
|
|
|
}
|
|
|
|
x := byTenantExpected[k]
|
|
|
|
if x == nil {
|
|
|
|
x = &uint64set.Set{}
|
|
|
|
byTenantExpected[k] = x
|
|
|
|
}
|
|
|
|
x.Add(e.MetricID)
|
|
|
|
}
|
2019-06-09 18:06:53 +02:00
|
|
|
|
|
|
|
hour := uint64(timestampFromTime(time.Now())) / msecPerHour
|
|
|
|
hmOrig := &hourMetricIDs{
|
2019-09-24 20:10:22 +02:00
|
|
|
m: &uint64set.Set{},
|
2019-06-09 18:06:53 +02:00
|
|
|
hour: 123,
|
|
|
|
}
|
2019-09-24 20:10:22 +02:00
|
|
|
hmOrig.m.Add(12)
|
|
|
|
hmOrig.m.Add(34)
|
2019-06-09 18:06:53 +02:00
|
|
|
s.currHourMetricIDs.Store(hmOrig)
|
|
|
|
s.updateCurrHourMetricIDs()
|
|
|
|
hmCurr := s.currHourMetricIDs.Load().(*hourMetricIDs)
|
|
|
|
if hmCurr.hour != hour {
|
2019-11-11 23:16:42 +01:00
|
|
|
// It is possible new hour occurred. Update the hour and verify it again.
|
2019-06-09 18:06:53 +02:00
|
|
|
hour = uint64(timestampFromTime(time.Now())) / msecPerHour
|
|
|
|
if hmCurr.hour != hour {
|
|
|
|
t.Fatalf("unexpected hmCurr.hour; got %d; want %d", hmCurr.hour, hour)
|
|
|
|
}
|
|
|
|
}
|
2019-11-08 12:16:40 +01:00
|
|
|
if !hmCurr.m.Equal(mExpected) {
|
2019-10-31 14:50:58 +01:00
|
|
|
t.Fatalf("unexpected hm.m; got %v; want %v", hmCurr.m, mExpected)
|
|
|
|
}
|
|
|
|
if !reflect.DeepEqual(hmCurr.byTenant, byTenantExpected) {
|
|
|
|
t.Fatalf("unexpected hmPrev.byTenant; got %v; want %v", hmCurr.byTenant, byTenantExpected)
|
2019-06-09 18:06:53 +02:00
|
|
|
}
|
|
|
|
if !hmCurr.isFull {
|
|
|
|
t.Fatalf("unexpected hmCurr.isFull; got %v; want %v", hmCurr.isFull, true)
|
|
|
|
}
|
|
|
|
|
|
|
|
hmPrev := s.prevHourMetricIDs.Load().(*hourMetricIDs)
|
|
|
|
if !reflect.DeepEqual(hmPrev, hmOrig) {
|
|
|
|
t.Fatalf("unexpected hmPrev; got %v; want %v", hmPrev, hmOrig)
|
|
|
|
}
|
2019-11-08 12:16:40 +01:00
|
|
|
if len(s.pendingHourEntries) != 0 {
|
|
|
|
t.Fatalf("unexpected len(s.pendingHourEntries); got %d; want %d", len(s.pendingHourEntries), 0)
|
2019-06-09 18:06:53 +02:00
|
|
|
}
|
|
|
|
})
|
|
|
|
t.Run("nonempty_pending_metric_ids_valid_curr_hour", func(t *testing.T) {
|
|
|
|
s := newStorage()
|
2019-11-08 12:16:40 +01:00
|
|
|
s.pendingHourEntries = []pendingHourMetricIDEntry{
|
2019-10-31 14:50:58 +01:00
|
|
|
{AccountID: 123, ProjectID: 431, MetricID: 343},
|
|
|
|
{AccountID: 123, ProjectID: 431, MetricID: 32424},
|
|
|
|
{AccountID: 1, ProjectID: 2, MetricID: 8293432},
|
|
|
|
}
|
|
|
|
mExpected := &uint64set.Set{}
|
2019-11-08 12:16:40 +01:00
|
|
|
for _, e := range s.pendingHourEntries {
|
2019-10-31 14:50:58 +01:00
|
|
|
mExpected.Add(e.MetricID)
|
|
|
|
}
|
|
|
|
byTenantExpected := make(map[accountProjectKey]*uint64set.Set)
|
2019-11-08 12:16:40 +01:00
|
|
|
for _, e := range s.pendingHourEntries {
|
2019-10-31 14:50:58 +01:00
|
|
|
k := accountProjectKey{
|
|
|
|
AccountID: e.AccountID,
|
|
|
|
ProjectID: e.ProjectID,
|
|
|
|
}
|
|
|
|
x := byTenantExpected[k]
|
|
|
|
if x == nil {
|
|
|
|
x = &uint64set.Set{}
|
|
|
|
byTenantExpected[k] = x
|
|
|
|
}
|
|
|
|
x.Add(e.MetricID)
|
|
|
|
}
|
2019-06-09 18:06:53 +02:00
|
|
|
|
|
|
|
hour := uint64(timestampFromTime(time.Now())) / msecPerHour
|
|
|
|
hmOrig := &hourMetricIDs{
|
2019-09-24 20:10:22 +02:00
|
|
|
m: &uint64set.Set{},
|
2019-06-09 18:06:53 +02:00
|
|
|
hour: hour,
|
|
|
|
}
|
2019-09-24 20:10:22 +02:00
|
|
|
hmOrig.m.Add(12)
|
|
|
|
hmOrig.m.Add(34)
|
2019-06-09 18:06:53 +02:00
|
|
|
s.currHourMetricIDs.Store(hmOrig)
|
|
|
|
s.updateCurrHourMetricIDs()
|
|
|
|
hmCurr := s.currHourMetricIDs.Load().(*hourMetricIDs)
|
|
|
|
if hmCurr.hour != hour {
|
2019-11-11 23:16:42 +01:00
|
|
|
// It is possible new hour occurred. Update the hour and verify it again.
|
2019-06-09 18:06:53 +02:00
|
|
|
hour = uint64(timestampFromTime(time.Now())) / msecPerHour
|
|
|
|
if hmCurr.hour != hour {
|
|
|
|
t.Fatalf("unexpected hmCurr.hour; got %d; want %d", hmCurr.hour, hour)
|
|
|
|
}
|
|
|
|
// Do not run other checks, since they may fail.
|
|
|
|
return
|
|
|
|
}
|
2019-10-31 14:50:58 +01:00
|
|
|
m := mExpected.Clone()
|
2020-01-15 11:12:46 +01:00
|
|
|
hmOrig.m.ForEach(func(part []uint64) bool {
|
|
|
|
for _, metricID := range part {
|
|
|
|
m.Add(metricID)
|
|
|
|
}
|
|
|
|
return true
|
|
|
|
})
|
2019-11-08 12:16:40 +01:00
|
|
|
if !hmCurr.m.Equal(m) {
|
2019-06-09 18:06:53 +02:00
|
|
|
t.Fatalf("unexpected hm.m; got %v; want %v", hmCurr.m, m)
|
|
|
|
}
|
2019-10-31 14:50:58 +01:00
|
|
|
if !reflect.DeepEqual(hmCurr.byTenant, byTenantExpected) {
|
|
|
|
t.Fatalf("unexpected hmPrev.byTenant; got %v; want %v", hmCurr.byTenant, byTenantExpected)
|
|
|
|
}
|
2019-06-09 18:06:53 +02:00
|
|
|
if hmCurr.isFull {
|
|
|
|
t.Fatalf("unexpected hmCurr.isFull; got %v; want %v", hmCurr.isFull, false)
|
|
|
|
}
|
|
|
|
|
|
|
|
hmPrev := s.prevHourMetricIDs.Load().(*hourMetricIDs)
|
|
|
|
hmEmpty := &hourMetricIDs{}
|
|
|
|
if !reflect.DeepEqual(hmPrev, hmEmpty) {
|
|
|
|
t.Fatalf("unexpected hmPrev; got %v; want %v", hmPrev, hmEmpty)
|
|
|
|
}
|
2019-11-08 12:16:40 +01:00
|
|
|
if len(s.pendingHourEntries) != 0 {
|
|
|
|
t.Fatalf("unexpected s.pendingHourEntries.Len(); got %d; want %d", len(s.pendingHourEntries), 0)
|
2019-06-09 18:06:53 +02:00
|
|
|
}
|
|
|
|
})
|
|
|
|
}
|
|
|
|
|
2019-05-22 23:16:55 +02:00
|
|
|
func TestMetricRowMarshalUnmarshal(t *testing.T) {
|
|
|
|
var buf []byte
|
|
|
|
typ := reflect.TypeOf(&MetricRow{})
|
|
|
|
rnd := rand.New(rand.NewSource(1))
|
|
|
|
|
|
|
|
for i := 0; i < 1000; i++ {
|
|
|
|
v, ok := quick.Value(typ, rnd)
|
|
|
|
if !ok {
|
|
|
|
t.Fatalf("cannot create random MetricRow via quick.Value")
|
|
|
|
}
|
|
|
|
mr1 := v.Interface().(*MetricRow)
|
|
|
|
if mr1 == nil {
|
|
|
|
continue
|
|
|
|
}
|
|
|
|
|
|
|
|
buf = mr1.Marshal(buf[:0])
|
|
|
|
var mr2 MetricRow
|
|
|
|
tail, err := mr2.Unmarshal(buf)
|
|
|
|
if err != nil {
|
|
|
|
t.Fatalf("cannot unmarshal mr1=%s: %s", mr1, err)
|
|
|
|
}
|
|
|
|
if len(tail) > 0 {
|
|
|
|
t.Fatalf("non-empty tail returned after MetricRow.Unmarshal for mr1=%s", mr1)
|
|
|
|
}
|
|
|
|
if mr1.MetricNameRaw == nil {
|
|
|
|
mr1.MetricNameRaw = []byte{}
|
|
|
|
}
|
|
|
|
if mr2.MetricNameRaw == nil {
|
|
|
|
mr2.MetricNameRaw = []byte{}
|
|
|
|
}
|
|
|
|
if !reflect.DeepEqual(mr1, &mr2) {
|
|
|
|
t.Fatalf("mr1 should match mr2; got\nmr1=%s\nmr2=%s", mr1, &mr2)
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
func TestNextRetentionDuration(t *testing.T) {
|
|
|
|
for retentionMonths := 1; retentionMonths < 360; retentionMonths++ {
|
|
|
|
d := nextRetentionDuration(retentionMonths)
|
2019-12-02 13:42:26 +01:00
|
|
|
if d <= 0 {
|
|
|
|
currTime := time.Now().UTC()
|
2019-05-22 23:16:55 +02:00
|
|
|
nextTime := time.Now().UTC().Add(d)
|
|
|
|
t.Fatalf("unexected retention duration for retentionMonths=%d; got %s; must be %s + %d months", retentionMonths, nextTime, currTime, retentionMonths)
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
func TestStorageOpenClose(t *testing.T) {
|
|
|
|
path := "TestStorageOpenClose"
|
|
|
|
for i := 0; i < 10; i++ {
|
|
|
|
s, err := OpenStorage(path, -1)
|
|
|
|
if err != nil {
|
|
|
|
t.Fatalf("cannot open storage: %s", err)
|
|
|
|
}
|
|
|
|
s.MustClose()
|
|
|
|
}
|
|
|
|
if err := os.RemoveAll(path); err != nil {
|
|
|
|
t.Fatalf("cannot remove %q: %s", path, err)
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
func TestStorageOpenMultipleTimes(t *testing.T) {
|
|
|
|
path := "TestStorageOpenMultipleTimes"
|
|
|
|
s1, err := OpenStorage(path, -1)
|
|
|
|
if err != nil {
|
|
|
|
t.Fatalf("cannot open storage the first time: %s", err)
|
|
|
|
}
|
|
|
|
|
|
|
|
for i := 0; i < 10; i++ {
|
|
|
|
s2, err := OpenStorage(path, -1)
|
|
|
|
if err == nil {
|
|
|
|
s2.MustClose()
|
|
|
|
t.Fatalf("expecting non-nil error when opening already opened storage")
|
|
|
|
}
|
|
|
|
}
|
|
|
|
s1.MustClose()
|
|
|
|
if err := os.RemoveAll(path); err != nil {
|
|
|
|
t.Fatalf("cannot remove %q: %s", path, err)
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
func TestStorageRandTimestamps(t *testing.T) {
|
|
|
|
path := "TestStorageRandTimestamps"
|
2020-10-20 13:29:26 +02:00
|
|
|
retentionMsecs := int64(60 * msecsPerMonth)
|
|
|
|
s, err := OpenStorage(path, retentionMsecs)
|
2019-05-22 23:16:55 +02:00
|
|
|
if err != nil {
|
|
|
|
t.Fatalf("cannot open storage: %s", err)
|
|
|
|
}
|
|
|
|
t.Run("serial", func(t *testing.T) {
|
|
|
|
for i := 0; i < 3; i++ {
|
|
|
|
if err := testStorageRandTimestamps(s); err != nil {
|
|
|
|
t.Fatal(err)
|
|
|
|
}
|
|
|
|
s.MustClose()
|
2020-10-20 13:29:26 +02:00
|
|
|
s, err = OpenStorage(path, retentionMsecs)
|
2019-05-22 23:16:55 +02:00
|
|
|
}
|
|
|
|
})
|
|
|
|
t.Run("concurrent", func(t *testing.T) {
|
|
|
|
ch := make(chan error, 3)
|
|
|
|
for i := 0; i < cap(ch); i++ {
|
|
|
|
go func() {
|
|
|
|
var err error
|
|
|
|
for i := 0; i < 2; i++ {
|
|
|
|
err = testStorageRandTimestamps(s)
|
|
|
|
}
|
|
|
|
ch <- err
|
|
|
|
}()
|
|
|
|
}
|
|
|
|
for i := 0; i < cap(ch); i++ {
|
|
|
|
select {
|
|
|
|
case err := <-ch:
|
|
|
|
if err != nil {
|
|
|
|
t.Fatal(err)
|
|
|
|
}
|
|
|
|
case <-time.After(time.Second * 10):
|
|
|
|
t.Fatal("timeout")
|
|
|
|
}
|
|
|
|
}
|
|
|
|
})
|
|
|
|
s.MustClose()
|
|
|
|
if err := os.RemoveAll(path); err != nil {
|
|
|
|
t.Fatalf("cannot remove %q: %s", path, err)
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
func testStorageRandTimestamps(s *Storage) error {
|
|
|
|
const rowsPerAdd = 1e3
|
|
|
|
const addsCount = 2
|
|
|
|
typ := reflect.TypeOf(int64(0))
|
|
|
|
rnd := rand.New(rand.NewSource(1))
|
|
|
|
|
|
|
|
for i := 0; i < addsCount; i++ {
|
|
|
|
var mrs []MetricRow
|
|
|
|
var mn MetricName
|
|
|
|
mn.Tags = []Tag{
|
|
|
|
{[]byte("job"), []byte("webservice")},
|
|
|
|
{[]byte("instance"), []byte("1.2.3.4")},
|
|
|
|
}
|
|
|
|
for j := 0; j < rowsPerAdd; j++ {
|
|
|
|
mn.MetricGroup = []byte(fmt.Sprintf("metric_%d", rand.Intn(100)))
|
|
|
|
metricNameRaw := mn.marshalRaw(nil)
|
|
|
|
timestamp := int64(rnd.NormFloat64() * 1e12)
|
|
|
|
if j%2 == 0 {
|
|
|
|
ts, ok := quick.Value(typ, rnd)
|
|
|
|
if !ok {
|
|
|
|
return fmt.Errorf("cannot create random timestamp via quick.Value")
|
|
|
|
}
|
|
|
|
timestamp = ts.Interface().(int64)
|
|
|
|
}
|
|
|
|
value := rnd.NormFloat64() * 1e12
|
|
|
|
|
|
|
|
mr := MetricRow{
|
|
|
|
MetricNameRaw: metricNameRaw,
|
|
|
|
Timestamp: timestamp,
|
|
|
|
Value: value,
|
|
|
|
}
|
|
|
|
mrs = append(mrs, mr)
|
|
|
|
}
|
|
|
|
if err := s.AddRows(mrs, defaultPrecisionBits); err != nil {
|
2019-08-25 14:28:32 +02:00
|
|
|
errStr := err.Error()
|
|
|
|
if !strings.Contains(errStr, "too big timestamp") && !strings.Contains(errStr, "too small timestamp") {
|
2020-06-30 21:58:18 +02:00
|
|
|
return fmt.Errorf("unexpected error when adding mrs: %w", err)
|
2019-05-22 23:16:55 +02:00
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
// Verify the storage contains rows.
|
|
|
|
var m Metrics
|
|
|
|
s.UpdateMetrics(&m)
|
|
|
|
if m.TableMetrics.SmallRowsCount == 0 {
|
|
|
|
return fmt.Errorf("expecting at least one row in the table")
|
|
|
|
}
|
|
|
|
return nil
|
|
|
|
}
|
|
|
|
|
|
|
|
func TestStorageDeleteMetrics(t *testing.T) {
|
|
|
|
path := "TestStorageDeleteMetrics"
|
|
|
|
s, err := OpenStorage(path, 0)
|
|
|
|
if err != nil {
|
|
|
|
t.Fatalf("cannot open storage: %s", err)
|
|
|
|
}
|
|
|
|
|
|
|
|
// Verify no tag keys exist
|
2020-07-23 19:42:57 +02:00
|
|
|
tks, err := s.SearchTagKeys(0, 0, 1e5, noDeadline)
|
2019-05-22 23:16:55 +02:00
|
|
|
if err != nil {
|
|
|
|
t.Fatalf("error in SearchTagKeys at the start: %s", err)
|
|
|
|
}
|
|
|
|
if len(tks) != 0 {
|
|
|
|
t.Fatalf("found non-empty tag keys at the start: %q", tks)
|
|
|
|
}
|
|
|
|
|
|
|
|
t.Run("serial", func(t *testing.T) {
|
|
|
|
for i := 0; i < 3; i++ {
|
|
|
|
if err = testStorageDeleteMetrics(s, 0); err != nil {
|
2019-11-09 22:17:42 +01:00
|
|
|
t.Fatalf("unexpected error on iteration %d: %s", i, err)
|
2019-05-22 23:16:55 +02:00
|
|
|
}
|
|
|
|
|
|
|
|
// Re-open the storage in order to check how deleted metricIDs
|
|
|
|
// are persisted.
|
|
|
|
s.MustClose()
|
|
|
|
s, err = OpenStorage(path, 0)
|
|
|
|
if err != nil {
|
2019-11-09 22:17:42 +01:00
|
|
|
t.Fatalf("cannot open storage after closing on iteration %d: %s", i, err)
|
2019-05-22 23:16:55 +02:00
|
|
|
}
|
|
|
|
}
|
|
|
|
})
|
|
|
|
|
|
|
|
t.Run("concurrent", func(t *testing.T) {
|
|
|
|
ch := make(chan error, 3)
|
|
|
|
for i := 0; i < cap(ch); i++ {
|
|
|
|
go func(workerNum int) {
|
|
|
|
var err error
|
|
|
|
for j := 0; j < 2; j++ {
|
|
|
|
err = testStorageDeleteMetrics(s, workerNum)
|
|
|
|
if err != nil {
|
|
|
|
break
|
|
|
|
}
|
|
|
|
}
|
|
|
|
ch <- err
|
|
|
|
}(i)
|
|
|
|
}
|
|
|
|
for i := 0; i < cap(ch); i++ {
|
|
|
|
select {
|
|
|
|
case err := <-ch:
|
|
|
|
if err != nil {
|
|
|
|
t.Fatalf("unexpected error: %s", err)
|
|
|
|
}
|
|
|
|
case <-time.After(30 * time.Second):
|
|
|
|
t.Fatalf("timeout")
|
|
|
|
}
|
|
|
|
}
|
|
|
|
})
|
|
|
|
|
|
|
|
// Verify no more tag keys exist
|
2020-07-23 19:42:57 +02:00
|
|
|
tks, err = s.SearchTagKeys(0, 0, 1e5, noDeadline)
|
2019-05-22 23:16:55 +02:00
|
|
|
if err != nil {
|
|
|
|
t.Fatalf("error in SearchTagKeys after the test: %s", err)
|
|
|
|
}
|
|
|
|
if len(tks) != 0 {
|
|
|
|
t.Fatalf("found non-empty tag keys after the test: %q", tks)
|
|
|
|
}
|
|
|
|
|
|
|
|
s.MustClose()
|
|
|
|
if err := os.RemoveAll(path); err != nil {
|
|
|
|
t.Fatalf("cannot remove %q: %s", path, err)
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
func testStorageDeleteMetrics(s *Storage, workerNum int) error {
|
|
|
|
const rowsPerMetric = 100
|
|
|
|
const metricsCount = 30
|
|
|
|
|
|
|
|
workerTag := []byte(fmt.Sprintf("workerTag_%d", workerNum))
|
2019-05-22 23:23:23 +02:00
|
|
|
accountID := uint32(workerNum)
|
|
|
|
projectID := uint32(123)
|
2019-05-22 23:16:55 +02:00
|
|
|
|
|
|
|
tksAll := make(map[string]bool)
|
|
|
|
tksAll[""] = true // __name__
|
|
|
|
for i := 0; i < metricsCount; i++ {
|
|
|
|
var mrs []MetricRow
|
|
|
|
var mn MetricName
|
2019-05-22 23:23:23 +02:00
|
|
|
mn.AccountID = accountID
|
|
|
|
mn.ProjectID = projectID
|
2019-05-22 23:16:55 +02:00
|
|
|
job := fmt.Sprintf("job_%d_%d", i, workerNum)
|
|
|
|
instance := fmt.Sprintf("instance_%d_%d", i, workerNum)
|
|
|
|
mn.Tags = []Tag{
|
|
|
|
{[]byte("job"), []byte(job)},
|
|
|
|
{[]byte("instance"), []byte(instance)},
|
|
|
|
{workerTag, []byte("foobar")},
|
|
|
|
}
|
|
|
|
for i := range mn.Tags {
|
|
|
|
tksAll[string(mn.Tags[i].Key)] = true
|
|
|
|
}
|
|
|
|
mn.MetricGroup = []byte(fmt.Sprintf("metric_%d_%d", i, workerNum))
|
|
|
|
metricNameRaw := mn.marshalRaw(nil)
|
|
|
|
|
|
|
|
for j := 0; j < rowsPerMetric; j++ {
|
|
|
|
timestamp := rand.Int63n(1e10)
|
|
|
|
value := rand.NormFloat64() * 1e6
|
|
|
|
|
|
|
|
mr := MetricRow{
|
|
|
|
MetricNameRaw: metricNameRaw,
|
|
|
|
Timestamp: timestamp,
|
|
|
|
Value: value,
|
|
|
|
}
|
|
|
|
mrs = append(mrs, mr)
|
|
|
|
}
|
|
|
|
if err := s.AddRows(mrs, defaultPrecisionBits); err != nil {
|
2020-06-30 21:58:18 +02:00
|
|
|
return fmt.Errorf("unexpected error when adding mrs: %w", err)
|
2019-05-22 23:16:55 +02:00
|
|
|
}
|
|
|
|
}
|
2020-11-11 13:40:27 +01:00
|
|
|
s.DebugFlush()
|
2019-05-22 23:16:55 +02:00
|
|
|
|
|
|
|
// Verify tag values exist
|
2020-07-23 19:42:57 +02:00
|
|
|
tvs, err := s.SearchTagValues(accountID, projectID, workerTag, 1e5, noDeadline)
|
2019-05-22 23:16:55 +02:00
|
|
|
if err != nil {
|
2020-06-30 21:58:18 +02:00
|
|
|
return fmt.Errorf("error in SearchTagValues before metrics removal: %w", err)
|
2019-05-22 23:16:55 +02:00
|
|
|
}
|
|
|
|
if len(tvs) == 0 {
|
|
|
|
return fmt.Errorf("unexpected empty number of tag values for workerTag")
|
|
|
|
}
|
|
|
|
|
|
|
|
// Verify tag keys exist
|
2020-07-23 19:42:57 +02:00
|
|
|
tks, err := s.SearchTagKeys(accountID, projectID, 1e5, noDeadline)
|
2019-05-22 23:16:55 +02:00
|
|
|
if err != nil {
|
2020-06-30 21:58:18 +02:00
|
|
|
return fmt.Errorf("error in SearchTagKeys before metrics removal: %w", err)
|
2019-05-22 23:16:55 +02:00
|
|
|
}
|
|
|
|
if err := checkTagKeys(tks, tksAll); err != nil {
|
2020-06-30 21:58:18 +02:00
|
|
|
return fmt.Errorf("unexpected tag keys before metrics removal: %w", err)
|
2019-05-22 23:16:55 +02:00
|
|
|
}
|
|
|
|
|
|
|
|
var sr Search
|
|
|
|
tr := TimeRange{
|
|
|
|
MinTimestamp: 0,
|
|
|
|
MaxTimestamp: 2e10,
|
|
|
|
}
|
|
|
|
metricBlocksCount := func(tfs *TagFilters) int {
|
2020-04-27 07:13:41 +02:00
|
|
|
// Verify the number of blocks
|
2019-05-22 23:16:55 +02:00
|
|
|
n := 0
|
2020-07-23 19:42:57 +02:00
|
|
|
sr.Init(s, []*TagFilters{tfs}, tr, 1e5, noDeadline)
|
2019-05-22 23:16:55 +02:00
|
|
|
for sr.NextMetricBlock() {
|
|
|
|
n++
|
|
|
|
}
|
|
|
|
sr.MustClose()
|
|
|
|
return n
|
|
|
|
}
|
|
|
|
for i := 0; i < metricsCount; i++ {
|
2019-05-22 23:23:23 +02:00
|
|
|
tfs := NewTagFilters(accountID, projectID)
|
2019-05-22 23:16:55 +02:00
|
|
|
if err := tfs.Add(nil, []byte("metric_.+"), false, true); err != nil {
|
2020-06-30 21:58:18 +02:00
|
|
|
return fmt.Errorf("cannot add regexp tag filter: %w", err)
|
2019-05-22 23:16:55 +02:00
|
|
|
}
|
|
|
|
job := fmt.Sprintf("job_%d_%d", i, workerNum)
|
|
|
|
if err := tfs.Add([]byte("job"), []byte(job), false, false); err != nil {
|
2020-06-30 21:58:18 +02:00
|
|
|
return fmt.Errorf("cannot add job tag filter: %w", err)
|
2019-05-22 23:16:55 +02:00
|
|
|
}
|
|
|
|
if n := metricBlocksCount(tfs); n == 0 {
|
|
|
|
return fmt.Errorf("expecting non-zero number of metric blocks for tfs=%s", tfs)
|
|
|
|
}
|
|
|
|
deletedCount, err := s.DeleteMetrics([]*TagFilters{tfs})
|
|
|
|
if err != nil {
|
2020-06-30 21:58:18 +02:00
|
|
|
return fmt.Errorf("cannot delete metrics: %w", err)
|
2019-05-22 23:16:55 +02:00
|
|
|
}
|
|
|
|
if deletedCount == 0 {
|
2019-11-09 17:48:58 +01:00
|
|
|
return fmt.Errorf("expecting non-zero number of deleted metrics on iteration %d", i)
|
2019-05-22 23:16:55 +02:00
|
|
|
}
|
|
|
|
if n := metricBlocksCount(tfs); n != 0 {
|
|
|
|
return fmt.Errorf("expecting zero metric blocks after DeleteMetrics call for tfs=%s; got %d blocks", tfs, n)
|
|
|
|
}
|
|
|
|
|
|
|
|
// Try deleting empty tfss
|
|
|
|
deletedCount, err = s.DeleteMetrics(nil)
|
|
|
|
if err != nil {
|
2020-06-30 21:58:18 +02:00
|
|
|
return fmt.Errorf("cannot delete empty tfss: %w", err)
|
2019-05-22 23:16:55 +02:00
|
|
|
}
|
|
|
|
if deletedCount != 0 {
|
|
|
|
return fmt.Errorf("expecting zero deleted metrics for empty tfss; got %d", deletedCount)
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
// Make sure no more metrics left for the given workerNum
|
2019-05-22 23:23:23 +02:00
|
|
|
tfs := NewTagFilters(accountID, projectID)
|
2019-05-22 23:16:55 +02:00
|
|
|
if err := tfs.Add(nil, []byte(fmt.Sprintf("metric_.+_%d", workerNum)), false, true); err != nil {
|
2020-06-30 21:58:18 +02:00
|
|
|
return fmt.Errorf("cannot add regexp tag filter for worker metrics: %w", err)
|
2019-05-22 23:16:55 +02:00
|
|
|
}
|
|
|
|
if n := metricBlocksCount(tfs); n != 0 {
|
|
|
|
return fmt.Errorf("expecting zero metric blocks after deleting all the metrics; got %d blocks", n)
|
|
|
|
}
|
2020-07-23 19:42:57 +02:00
|
|
|
tvs, err = s.SearchTagValues(accountID, projectID, workerTag, 1e5, noDeadline)
|
2019-05-22 23:16:55 +02:00
|
|
|
if err != nil {
|
2020-06-30 21:58:18 +02:00
|
|
|
return fmt.Errorf("error in SearchTagValues after all the metrics are removed: %w", err)
|
2019-05-22 23:16:55 +02:00
|
|
|
}
|
|
|
|
if len(tvs) != 0 {
|
|
|
|
return fmt.Errorf("found non-empty tag values for %q after metrics removal: %q", workerTag, tvs)
|
|
|
|
}
|
|
|
|
|
|
|
|
return nil
|
|
|
|
}
|
|
|
|
|
|
|
|
func checkTagKeys(tks []string, tksExpected map[string]bool) error {
|
|
|
|
if len(tks) < len(tksExpected) {
|
|
|
|
return fmt.Errorf("unexpected number of tag keys found; got %d; want at least %d; tks=%q, tksExpected=%v", len(tks), len(tksExpected), tks, tksExpected)
|
|
|
|
}
|
|
|
|
hasItem := func(k string, tks []string) bool {
|
|
|
|
for _, kk := range tks {
|
|
|
|
if k == kk {
|
|
|
|
return true
|
|
|
|
}
|
|
|
|
}
|
|
|
|
return false
|
|
|
|
}
|
|
|
|
for k := range tksExpected {
|
|
|
|
if !hasItem(k, tks) {
|
|
|
|
return fmt.Errorf("cannot find %q in tag keys %q", k, tks)
|
|
|
|
}
|
|
|
|
}
|
|
|
|
return nil
|
|
|
|
}
|
|
|
|
|
2020-11-15 23:42:27 +01:00
|
|
|
func TestStorageRegisterMetricNamesSerial(t *testing.T) {
|
|
|
|
path := "TestStorageRegisterMetricNamesSerial"
|
|
|
|
s, err := OpenStorage(path, 0)
|
|
|
|
if err != nil {
|
|
|
|
t.Fatalf("cannot open storage: %s", err)
|
|
|
|
}
|
|
|
|
if err := testStorageRegisterMetricNames(s); err != nil {
|
|
|
|
t.Fatalf("unexpected error: %s", err)
|
|
|
|
}
|
|
|
|
s.MustClose()
|
|
|
|
if err := os.RemoveAll(path); err != nil {
|
|
|
|
t.Fatalf("cannot remove %q: %s", path, err)
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
func TestStorageRegisterMetricNamesConcurrent(t *testing.T) {
|
|
|
|
path := "TestStorageRegisterMetricNamesConcurrent"
|
|
|
|
s, err := OpenStorage(path, 0)
|
|
|
|
if err != nil {
|
|
|
|
t.Fatalf("cannot open storage: %s", err)
|
|
|
|
}
|
|
|
|
ch := make(chan error, 3)
|
|
|
|
for i := 0; i < cap(ch); i++ {
|
|
|
|
go func() {
|
|
|
|
ch <- testStorageRegisterMetricNames(s)
|
|
|
|
}()
|
|
|
|
}
|
|
|
|
for i := 0; i < cap(ch); i++ {
|
|
|
|
select {
|
|
|
|
case err := <-ch:
|
|
|
|
if err != nil {
|
|
|
|
t.Fatalf("unexpected error: %s", err)
|
|
|
|
}
|
|
|
|
case <-time.After(10 * time.Second):
|
|
|
|
t.Fatalf("timeout")
|
|
|
|
}
|
|
|
|
}
|
|
|
|
s.MustClose()
|
|
|
|
if err := os.RemoveAll(path); err != nil {
|
|
|
|
t.Fatalf("cannot remove %q: %s", path, err)
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
func testStorageRegisterMetricNames(s *Storage) error {
|
|
|
|
const metricsPerAdd = 1e3
|
|
|
|
const addsCount = 10
|
|
|
|
const accountID = 123
|
|
|
|
const projectID = 421
|
|
|
|
|
|
|
|
addIDsMap := make(map[string]struct{})
|
|
|
|
for i := 0; i < addsCount; i++ {
|
|
|
|
var mrs []MetricRow
|
|
|
|
var mn MetricName
|
|
|
|
addID := fmt.Sprintf("%d", i)
|
|
|
|
addIDsMap[addID] = struct{}{}
|
|
|
|
mn.AccountID = accountID
|
|
|
|
mn.ProjectID = projectID
|
|
|
|
mn.Tags = []Tag{
|
|
|
|
{[]byte("job"), []byte("webservice")},
|
|
|
|
{[]byte("instance"), []byte("1.2.3.4")},
|
|
|
|
{[]byte("add_id"), []byte(addID)},
|
|
|
|
}
|
|
|
|
now := timestampFromTime(time.Now())
|
|
|
|
for j := 0; j < metricsPerAdd; j++ {
|
2020-11-16 12:15:16 +01:00
|
|
|
mn.MetricGroup = []byte(fmt.Sprintf("metric_%d", j))
|
2020-11-15 23:42:27 +01:00
|
|
|
metricNameRaw := mn.marshalRaw(nil)
|
|
|
|
|
|
|
|
mr := MetricRow{
|
|
|
|
MetricNameRaw: metricNameRaw,
|
|
|
|
Timestamp: now,
|
|
|
|
}
|
|
|
|
mrs = append(mrs, mr)
|
|
|
|
}
|
|
|
|
if err := s.RegisterMetricNames(mrs); err != nil {
|
|
|
|
return fmt.Errorf("unexpected error in AddMetrics: %w", err)
|
|
|
|
}
|
|
|
|
}
|
|
|
|
var addIDsExpected []string
|
|
|
|
for k := range addIDsMap {
|
|
|
|
addIDsExpected = append(addIDsExpected, k)
|
|
|
|
}
|
|
|
|
sort.Strings(addIDsExpected)
|
|
|
|
|
|
|
|
// Verify the storage contains the added metric names.
|
|
|
|
s.DebugFlush()
|
|
|
|
|
|
|
|
// Verify that SearchTagKeys returns correct result.
|
|
|
|
tksExpected := []string{
|
|
|
|
"",
|
|
|
|
"add_id",
|
|
|
|
"instance",
|
|
|
|
"job",
|
|
|
|
}
|
|
|
|
tks, err := s.SearchTagKeys(accountID, projectID, 100, noDeadline)
|
|
|
|
if err != nil {
|
|
|
|
return fmt.Errorf("error in SearchTagKeys: %w", err)
|
|
|
|
}
|
|
|
|
sort.Strings(tks)
|
|
|
|
if !reflect.DeepEqual(tks, tksExpected) {
|
|
|
|
return fmt.Errorf("unexpected tag keys returned from SearchTagKeys;\ngot\n%q\nwant\n%q", tks, tksExpected)
|
|
|
|
}
|
|
|
|
|
2020-11-16 16:59:01 +01:00
|
|
|
// Verify that SearchTagKeys returns empty results for incorrect accountID, projectID
|
|
|
|
tks, err = s.SearchTagKeys(accountID+1, projectID+1, 100, noDeadline)
|
|
|
|
if err != nil {
|
|
|
|
return fmt.Errorf("error in SearchTagKeys for incorrect accountID, projectID: %w", err)
|
|
|
|
}
|
|
|
|
if len(tks) > 0 {
|
|
|
|
return fmt.Errorf("SearchTagKeys with incorrect accountID, projectID returns unexpected non-empty result:\n%q", tks)
|
|
|
|
}
|
|
|
|
|
2020-11-15 23:42:27 +01:00
|
|
|
// Verify that SearchTagKeysOnTimeRange returns correct result.
|
|
|
|
now := timestampFromTime(time.Now())
|
|
|
|
start := now - msecPerDay
|
|
|
|
end := now + 60*1000
|
|
|
|
tr := TimeRange{
|
|
|
|
MinTimestamp: start,
|
|
|
|
MaxTimestamp: end,
|
|
|
|
}
|
|
|
|
tks, err = s.SearchTagKeysOnTimeRange(accountID, projectID, tr, 100, noDeadline)
|
|
|
|
if err != nil {
|
|
|
|
return fmt.Errorf("error in SearchTagKeysOnTimeRange: %w", err)
|
|
|
|
}
|
|
|
|
sort.Strings(tks)
|
|
|
|
if !reflect.DeepEqual(tks, tksExpected) {
|
|
|
|
return fmt.Errorf("unexpected tag keys returned from SearchTagKeysOnTimeRange;\ngot\n%q\nwant\n%q", tks, tksExpected)
|
|
|
|
}
|
|
|
|
|
2020-11-16 16:59:01 +01:00
|
|
|
// Verify that SearchTagKeysOnTimeRange returns empty results for incrorrect accountID, projectID
|
|
|
|
tks, err = s.SearchTagKeysOnTimeRange(accountID+1, projectID+1, tr, 100, noDeadline)
|
|
|
|
if err != nil {
|
|
|
|
return fmt.Errorf("error in SearchTagKeysOnTimeRange for incorrect accountID, projectID: %w", err)
|
|
|
|
}
|
|
|
|
if len(tks) > 0 {
|
|
|
|
return fmt.Errorf("SearchTagKeysOnTimeRange with incorrect accountID, projectID returns unexpected non-empty result:\n%q", tks)
|
|
|
|
}
|
|
|
|
|
2020-11-15 23:42:27 +01:00
|
|
|
// Verify that SearchTagValues returns correct result.
|
|
|
|
addIDs, err := s.SearchTagValues(accountID, projectID, []byte("add_id"), addsCount+100, noDeadline)
|
|
|
|
if err != nil {
|
|
|
|
return fmt.Errorf("error in SearchTagValues: %w", err)
|
|
|
|
}
|
|
|
|
sort.Strings(addIDs)
|
|
|
|
if !reflect.DeepEqual(addIDs, addIDsExpected) {
|
|
|
|
return fmt.Errorf("unexpected tag values returned from SearchTagValues;\ngot\n%q\nwant\n%q", addIDs, addIDsExpected)
|
|
|
|
}
|
|
|
|
|
2020-11-16 16:59:01 +01:00
|
|
|
// Verify that SearchTagValues return empty results for incorrect accountID, projectID
|
|
|
|
addIDs, err = s.SearchTagValues(accountID+1, projectID+1, []byte("add_id"), addsCount+100, noDeadline)
|
|
|
|
if err != nil {
|
|
|
|
return fmt.Errorf("error in SearchTagValues for incorrect accountID, projectID: %w", err)
|
|
|
|
}
|
|
|
|
if len(addIDs) > 0 {
|
|
|
|
return fmt.Errorf("SearchTagValues with incorrect accountID, projectID returns unexpected non-empty result:\n%q", addIDs)
|
|
|
|
}
|
|
|
|
|
2020-11-15 23:42:27 +01:00
|
|
|
// Verify that SearchTagValuesOnTimeRange returns correct result.
|
|
|
|
addIDs, err = s.SearchTagValuesOnTimeRange(accountID, projectID, []byte("add_id"), tr, addsCount+100, noDeadline)
|
|
|
|
if err != nil {
|
|
|
|
return fmt.Errorf("error in SearchTagValuesOnTimeRange: %w", err)
|
|
|
|
}
|
|
|
|
sort.Strings(addIDs)
|
|
|
|
if !reflect.DeepEqual(addIDs, addIDsExpected) {
|
|
|
|
return fmt.Errorf("unexpected tag values returned from SearchTagValuesOnTimeRange;\ngot\n%q\nwant\n%q", addIDs, addIDsExpected)
|
|
|
|
}
|
|
|
|
|
2020-11-16 16:59:01 +01:00
|
|
|
// Verify that SearchTagValuesOnTimeRange returns empty results for incorrect accountID, projectID
|
|
|
|
addIDs, err = s.SearchTagValuesOnTimeRange(accountID+1, projectID+1, []byte("addd_id"), tr, addsCount+100, noDeadline)
|
|
|
|
if err != nil {
|
|
|
|
return fmt.Errorf("error in SearchTagValuesOnTimeRange for incorrect accoundID, projectID: %w", err)
|
|
|
|
}
|
|
|
|
if len(addIDs) > 0 {
|
|
|
|
return fmt.Errorf("SearchTagValuesOnTimeRange with incorrect accountID, projectID returns unexpected non-empty result:\n%q", addIDs)
|
|
|
|
}
|
|
|
|
|
2020-11-16 12:15:16 +01:00
|
|
|
// Verify that SearchMetricNames returns correct result.
|
|
|
|
tfs := NewTagFilters(accountID, projectID)
|
|
|
|
if err := tfs.Add([]byte("add_id"), []byte("0"), false, false); err != nil {
|
|
|
|
return fmt.Errorf("unexpected error in TagFilters.Add: %w", err)
|
|
|
|
}
|
2020-11-16 16:59:01 +01:00
|
|
|
mns, err := s.SearchMetricNames([]*TagFilters{tfs}, tr, metricsPerAdd*addsCount*100+100, noDeadline)
|
2020-11-16 12:15:16 +01:00
|
|
|
if err != nil {
|
|
|
|
return fmt.Errorf("error in SearchMetricNames: %w", err)
|
|
|
|
}
|
|
|
|
if len(mns) < metricsPerAdd {
|
|
|
|
return fmt.Errorf("unexpected number of metricNames returned from SearchMetricNames; got %d; want at least %d", len(mns), int(metricsPerAdd))
|
|
|
|
}
|
|
|
|
for i, mn := range mns {
|
|
|
|
addID := mn.GetTagValue("add_id")
|
|
|
|
if string(addID) != "0" {
|
|
|
|
return fmt.Errorf("unexpected addID for metricName #%d; got %q; want %q", i, addID, "0")
|
|
|
|
}
|
|
|
|
job := mn.GetTagValue("job")
|
|
|
|
if string(job) != "webservice" {
|
|
|
|
return fmt.Errorf("unexpected job for metricName #%d; got %q; want %q", i, job, "webservice")
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2020-11-16 16:59:01 +01:00
|
|
|
// Verify that SearchMetricNames returns empty results for incorrect accountID, projectID
|
|
|
|
tfs = NewTagFilters(accountID+1, projectID+1)
|
|
|
|
if err := tfs.Add([]byte("add_id"), []byte("0"), false, false); err != nil {
|
|
|
|
return fmt.Errorf("unexpected error in TagFilters.Add: %w", err)
|
|
|
|
}
|
|
|
|
mns, err = s.SearchMetricNames([]*TagFilters{tfs}, tr, metricsPerAdd*addsCount*100+100, noDeadline)
|
|
|
|
if err != nil {
|
|
|
|
return fmt.Errorf("error in SearchMetricNames for incorrect accountID, projectID: %w", err)
|
|
|
|
}
|
|
|
|
if len(mns) > 0 {
|
|
|
|
return fmt.Errorf("SearchMetricNames with incorrect accountID, projectID returns unexpected non-empty result:\n%+v", mns)
|
|
|
|
}
|
|
|
|
|
2020-11-15 23:42:27 +01:00
|
|
|
return nil
|
|
|
|
}
|
|
|
|
|
2020-03-24 21:24:54 +01:00
|
|
|
func TestStorageAddRowsSerial(t *testing.T) {
|
|
|
|
path := "TestStorageAddRowsSerial"
|
2019-05-22 23:16:55 +02:00
|
|
|
s, err := OpenStorage(path, 0)
|
|
|
|
if err != nil {
|
|
|
|
t.Fatalf("cannot open storage: %s", err)
|
|
|
|
}
|
2020-03-24 21:24:54 +01:00
|
|
|
if err := testStorageAddRows(s); err != nil {
|
|
|
|
t.Fatalf("unexpected error: %s", err)
|
|
|
|
}
|
|
|
|
s.MustClose()
|
|
|
|
if err := os.RemoveAll(path); err != nil {
|
|
|
|
t.Fatalf("cannot remove %q: %s", path, err)
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
func TestStorageAddRowsConcurrent(t *testing.T) {
|
|
|
|
path := "TestStorageAddRowsConcurrent"
|
|
|
|
s, err := OpenStorage(path, 0)
|
|
|
|
if err != nil {
|
|
|
|
t.Fatalf("cannot open storage: %s", err)
|
|
|
|
}
|
|
|
|
ch := make(chan error, 3)
|
|
|
|
for i := 0; i < cap(ch); i++ {
|
|
|
|
go func() {
|
|
|
|
ch <- testStorageAddRows(s)
|
|
|
|
}()
|
|
|
|
}
|
|
|
|
for i := 0; i < cap(ch); i++ {
|
|
|
|
select {
|
|
|
|
case err := <-ch:
|
|
|
|
if err != nil {
|
|
|
|
t.Fatalf("unexpected error: %s", err)
|
2019-05-22 23:16:55 +02:00
|
|
|
}
|
2020-03-24 21:24:54 +01:00
|
|
|
case <-time.After(10 * time.Second):
|
|
|
|
t.Fatalf("timeout")
|
2019-05-22 23:16:55 +02:00
|
|
|
}
|
2020-03-24 21:24:54 +01:00
|
|
|
}
|
2019-05-22 23:16:55 +02:00
|
|
|
s.MustClose()
|
|
|
|
if err := os.RemoveAll(path); err != nil {
|
|
|
|
t.Fatalf("cannot remove %q: %s", path, err)
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
func testStorageAddRows(s *Storage) error {
|
|
|
|
const rowsPerAdd = 1e3
|
|
|
|
const addsCount = 10
|
|
|
|
|
|
|
|
for i := 0; i < addsCount; i++ {
|
|
|
|
var mrs []MetricRow
|
|
|
|
var mn MetricName
|
|
|
|
mn.Tags = []Tag{
|
|
|
|
{[]byte("job"), []byte("webservice")},
|
|
|
|
{[]byte("instance"), []byte("1.2.3.4")},
|
|
|
|
}
|
|
|
|
for j := 0; j < rowsPerAdd; j++ {
|
2019-05-22 23:23:23 +02:00
|
|
|
mn.AccountID = uint32(rand.Intn(2))
|
|
|
|
mn.ProjectID = uint32(rand.Intn(3))
|
2019-05-22 23:16:55 +02:00
|
|
|
mn.MetricGroup = []byte(fmt.Sprintf("metric_%d", rand.Intn(100)))
|
|
|
|
metricNameRaw := mn.marshalRaw(nil)
|
|
|
|
timestamp := rand.Int63n(1e10)
|
|
|
|
value := rand.NormFloat64() * 1e6
|
|
|
|
|
|
|
|
mr := MetricRow{
|
|
|
|
MetricNameRaw: metricNameRaw,
|
|
|
|
Timestamp: timestamp,
|
|
|
|
Value: value,
|
|
|
|
}
|
|
|
|
mrs = append(mrs, mr)
|
|
|
|
}
|
|
|
|
if err := s.AddRows(mrs, defaultPrecisionBits); err != nil {
|
2020-06-30 21:58:18 +02:00
|
|
|
return fmt.Errorf("unexpected error when adding mrs: %w", err)
|
2019-05-22 23:16:55 +02:00
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
// Verify the storage contains rows.
|
|
|
|
minRowsExpected := uint64(rowsPerAdd) * addsCount
|
|
|
|
var m Metrics
|
|
|
|
s.UpdateMetrics(&m)
|
|
|
|
if m.TableMetrics.SmallRowsCount < minRowsExpected {
|
|
|
|
return fmt.Errorf("expecting at least %d rows in the table; got %d", minRowsExpected, m.TableMetrics.SmallRowsCount)
|
|
|
|
}
|
|
|
|
|
|
|
|
// Try creating a snapshot from the storage.
|
|
|
|
snapshotName, err := s.CreateSnapshot()
|
|
|
|
if err != nil {
|
2020-06-30 21:58:18 +02:00
|
|
|
return fmt.Errorf("cannot create snapshot from the storage: %w", err)
|
2019-05-22 23:16:55 +02:00
|
|
|
}
|
|
|
|
|
|
|
|
// Verify the snapshot is visible
|
|
|
|
snapshots, err := s.ListSnapshots()
|
|
|
|
if err != nil {
|
2020-06-30 21:58:18 +02:00
|
|
|
return fmt.Errorf("cannot list snapshots: %w", err)
|
2019-05-22 23:16:55 +02:00
|
|
|
}
|
|
|
|
if !containsString(snapshots, snapshotName) {
|
|
|
|
return fmt.Errorf("cannot find snapshot %q in %q", snapshotName, snapshots)
|
|
|
|
}
|
|
|
|
|
|
|
|
// Try opening the storage from snapshot.
|
|
|
|
snapshotPath := s.path + "/snapshots/" + snapshotName
|
|
|
|
s1, err := OpenStorage(snapshotPath, 0)
|
|
|
|
if err != nil {
|
2020-06-30 21:58:18 +02:00
|
|
|
return fmt.Errorf("cannot open storage from snapshot: %w", err)
|
2019-05-22 23:16:55 +02:00
|
|
|
}
|
|
|
|
|
|
|
|
// Verify the snapshot contains rows
|
|
|
|
var m1 Metrics
|
|
|
|
s1.UpdateMetrics(&m1)
|
|
|
|
if m1.TableMetrics.SmallRowsCount < minRowsExpected {
|
|
|
|
return fmt.Errorf("snapshot %q must contain at least %d rows; got %d", snapshotPath, minRowsExpected, m1.TableMetrics.SmallRowsCount)
|
|
|
|
}
|
|
|
|
|
2020-09-17 11:01:53 +02:00
|
|
|
// Verify that force merge for the snapshot leaves only a single part per partition.
|
|
|
|
if err := s1.ForceMergePartitions(""); err != nil {
|
|
|
|
return fmt.Errorf("error when force merging partitions: %w", err)
|
|
|
|
}
|
|
|
|
ptws := s1.tb.GetPartitions(nil)
|
|
|
|
defer s1.tb.PutPartitions(ptws)
|
|
|
|
for _, ptw := range ptws {
|
|
|
|
pws := ptw.pt.GetParts(nil)
|
|
|
|
numParts := len(pws)
|
|
|
|
ptw.pt.PutParts(pws)
|
|
|
|
if numParts != 1 {
|
|
|
|
return fmt.Errorf("unexpected number of parts for partition %q after force merge; got %d; want 1", ptw.pt.name, numParts)
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2019-05-22 23:16:55 +02:00
|
|
|
s1.MustClose()
|
|
|
|
|
|
|
|
// Delete the snapshot and make sure it is no longer visible.
|
|
|
|
if err := s.DeleteSnapshot(snapshotName); err != nil {
|
2020-06-30 21:58:18 +02:00
|
|
|
return fmt.Errorf("cannot delete snapshot %q: %w", snapshotName, err)
|
2019-05-22 23:16:55 +02:00
|
|
|
}
|
|
|
|
snapshots, err = s.ListSnapshots()
|
|
|
|
if err != nil {
|
2020-06-30 21:58:18 +02:00
|
|
|
return fmt.Errorf("cannot list snapshots: %w", err)
|
2019-05-22 23:16:55 +02:00
|
|
|
}
|
|
|
|
if containsString(snapshots, snapshotName) {
|
|
|
|
return fmt.Errorf("snapshot %q must be deleted, but is still visible in %q", snapshotName, snapshots)
|
|
|
|
}
|
|
|
|
|
|
|
|
return nil
|
|
|
|
}
|
|
|
|
|
|
|
|
func TestStorageRotateIndexDB(t *testing.T) {
|
|
|
|
path := "TestStorageRotateIndexDB"
|
|
|
|
s, err := OpenStorage(path, 0)
|
|
|
|
if err != nil {
|
|
|
|
t.Fatalf("cannot open storage: %s", err)
|
|
|
|
}
|
|
|
|
|
|
|
|
// Start indexDB rotater in a separate goroutine
|
|
|
|
stopCh := make(chan struct{})
|
|
|
|
rotateDoneCh := make(chan struct{})
|
|
|
|
go func() {
|
|
|
|
for {
|
|
|
|
select {
|
|
|
|
case <-stopCh:
|
|
|
|
close(rotateDoneCh)
|
|
|
|
return
|
|
|
|
default:
|
|
|
|
time.Sleep(time.Millisecond)
|
|
|
|
s.mustRotateIndexDB()
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}()
|
|
|
|
|
|
|
|
// Run concurrent workers that insert / select data from the storage.
|
|
|
|
ch := make(chan error, 3)
|
|
|
|
for i := 0; i < cap(ch); i++ {
|
|
|
|
go func(workerNum int) {
|
|
|
|
ch <- testStorageAddMetrics(s, workerNum)
|
|
|
|
}(i)
|
|
|
|
}
|
|
|
|
for i := 0; i < cap(ch); i++ {
|
|
|
|
select {
|
|
|
|
case err := <-ch:
|
|
|
|
if err != nil {
|
|
|
|
t.Fatalf("unexpected error: %s", err)
|
|
|
|
}
|
|
|
|
case <-time.After(10 * time.Second):
|
|
|
|
t.Fatalf("timeout")
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
close(stopCh)
|
|
|
|
<-rotateDoneCh
|
|
|
|
|
|
|
|
s.MustClose()
|
|
|
|
if err := os.RemoveAll(path); err != nil {
|
|
|
|
t.Fatalf("cannot remove %q: %s", path, err)
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
func testStorageAddMetrics(s *Storage, workerNum int) error {
|
|
|
|
const rowsCount = 1e3
|
|
|
|
|
|
|
|
var mn MetricName
|
|
|
|
mn.Tags = []Tag{
|
|
|
|
{[]byte("job"), []byte(fmt.Sprintf("webservice_%d", workerNum))},
|
|
|
|
{[]byte("instance"), []byte("1.2.3.4")},
|
|
|
|
}
|
|
|
|
for i := 0; i < rowsCount; i++ {
|
2019-05-22 23:23:23 +02:00
|
|
|
mn.AccountID = 123
|
|
|
|
mn.ProjectID = uint32(i % 3)
|
2019-05-22 23:16:55 +02:00
|
|
|
mn.MetricGroup = []byte(fmt.Sprintf("metric_%d_%d", workerNum, rand.Intn(10)))
|
|
|
|
metricNameRaw := mn.marshalRaw(nil)
|
|
|
|
timestamp := rand.Int63n(1e10)
|
|
|
|
value := rand.NormFloat64() * 1e6
|
|
|
|
|
|
|
|
mr := MetricRow{
|
|
|
|
MetricNameRaw: metricNameRaw,
|
|
|
|
Timestamp: timestamp,
|
|
|
|
Value: value,
|
|
|
|
}
|
|
|
|
if err := s.AddRows([]MetricRow{mr}, defaultPrecisionBits); err != nil {
|
2020-06-30 21:58:18 +02:00
|
|
|
return fmt.Errorf("unexpected error when adding mrs: %w", err)
|
2019-05-22 23:16:55 +02:00
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
// Verify the storage contains rows.
|
|
|
|
minRowsExpected := uint64(rowsCount)
|
|
|
|
var m Metrics
|
|
|
|
s.UpdateMetrics(&m)
|
|
|
|
if m.TableMetrics.SmallRowsCount < minRowsExpected {
|
|
|
|
return fmt.Errorf("expecting at least %d rows in the table; got %d", minRowsExpected, m.TableMetrics.SmallRowsCount)
|
|
|
|
}
|
|
|
|
return nil
|
|
|
|
}
|
|
|
|
|
|
|
|
func containsString(a []string, s string) bool {
|
|
|
|
for i := range a {
|
|
|
|
if a[i] == s {
|
|
|
|
return true
|
|
|
|
}
|
|
|
|
}
|
|
|
|
return false
|
|
|
|
}
|