You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@skywalking.apache.org by ha...@apache.org on 2024/05/03 13:03:18 UTC
(skywalking-banyandb) branch long-running updated: Fix rotation issues
This is an automated email from the ASF dual-hosted git repository.
hanahmily pushed a commit to branch long-running
in repository https://gitbox.apache.org/repos/asf/skywalking-banyandb.git
The following commit(s) were added to refs/heads/long-running by this push:
new 45dfd167 Fix rotation issues
45dfd167 is described below
commit 45dfd167d1bed66461857ab5e2d608aaa64579e7
Author: Gao Hongtao <ha...@gmail.com>
AuthorDate: Fri May 3 12:55:53 2024 +0000
Fix rotation issues
Signed-off-by: Gao Hongtao <ha...@gmail.com>
---
banyand/internal/storage/index.go | 89 +++++++++-------
banyand/internal/storage/rotation.go | 8 +-
banyand/internal/storage/rotation_test.go | 4 +-
banyand/internal/storage/segment.go | 5 +-
banyand/internal/storage/tsdb.go | 22 ++--
banyand/stream/elements.go | 8 +-
banyand/stream/write.go | 4 +
pkg/test/flags/flags.go | 1 +
pkg/test/helpers/file.go | 63 ++++++++++++
pkg/test/setup/setup.go | 2 +-
pkg/test/space.go | 21 ++++
pkg/test/stream/testdata/group.json | 2 +-
pkg/timestamp/range.go | 1 +
test/integration/load/load_suite_test.go | 145 ++++++++++++++++++---------
test/stress/istio/istio_suite_test.go | 39 +------
test/stress/trace/docker-compose-single.yaml | 1 +
test/stress/trace/env | 2 +-
test/stress/trace/segment.tpl.json | 5 +
18 files changed, 270 insertions(+), 152 deletions(-)
diff --git a/banyand/internal/storage/index.go b/banyand/internal/storage/index.go
index 1c60c8f8..62f277de 100644
--- a/banyand/internal/storage/index.go
+++ b/banyand/internal/storage/index.go
@@ -275,7 +275,7 @@ func newSeriesIndexController[T TSTable, O any](
}
switch len(idxName) {
case 0:
- if sic.hot, err = sic.newIdx(ctx); err != nil {
+ if sic.hot, err = sic.newIdx(ctx, clock.Now()); err != nil {
return nil, err
}
case 1:
@@ -317,13 +317,13 @@ func (sic *seriesIndexController[T, O]) loadIdx() ([]string, error) {
return idxName, nil
}
-func (sic *seriesIndexController[T, O]) newIdx(ctx context.Context) (*seriesIndex, error) {
- ts := sic.opts.TTL.Unit.standard(sic.clock.Now())
+func (sic *seriesIndexController[T, O]) newIdx(ctx context.Context, now time.Time) (*seriesIndex, error) {
+ ts := sic.opts.TTL.Unit.standard(now)
return sic.openIdx(ctx, fmt.Sprintf("idx-%016x", ts.UnixNano()))
}
-func (sic *seriesIndexController[T, O]) newNextIdx(ctx context.Context) (*seriesIndex, error) {
- ts := sic.opts.TTL.Unit.standard(sic.clock.Now())
+func (sic *seriesIndexController[T, O]) newNextIdx(ctx context.Context, now time.Time) (*seriesIndex, error) {
+ ts := sic.opts.TTL.Unit.standard(now)
ts = ts.Add(sic.standbyLiveTime)
return sic.openIdx(ctx, fmt.Sprintf("idx-%016x", ts.UnixNano()))
}
@@ -341,47 +341,66 @@ func (sic *seriesIndexController[T, O]) openIdx(ctx context.Context, name string
return nil, errors.New("unexpected series index name")
}
-func (sic *seriesIndexController[T, O]) run(deadline time.Time) (err error) {
- var standby *seriesIndex
+func (sic *seriesIndexController[T, O]) run(now, deadline time.Time) (err error) {
ctx := context.WithValue(context.Background(), logger.ContextKey, sic.l)
- _, err = sic.loadIdx()
- if err != nil {
+ if _, err := sic.loadIdx(); err != nil {
sic.l.Warn().Err(err).Msg("fail to clear redundant series index")
}
+
+ sic.Lock()
+ defer sic.Unlock()
if sic.hot.startTime.Compare(deadline) <= 0 {
- sic.l.Info().Time("deadline", deadline).Msg("start to swap series index")
- sic.Lock()
- if sic.standby == nil {
- sic.standby, err = sic.newIdx(ctx)
- if err != nil {
- sic.Unlock()
- return err
- }
- }
- standby = sic.hot
- sic.hot = sic.standby
- sic.standby = nil
- sic.Unlock()
- err = standby.Close()
- if err != nil {
- sic.l.Warn().Err(err).Msg("fail to close standby series index")
- }
- lfs.MustRMAll(standby.path)
- sic.l.Info().Str("path", standby.path).Msg("dropped series index")
- lfs.SyncPath(sic.location)
+ return sic.handleStandby(ctx, now, deadline)
}
+ if sic.standby != nil {
+ return nil
+ }
liveTime := sic.hot.startTime.Sub(deadline)
- if liveTime > 0 && liveTime <= sic.standbyLiveTime {
- sic.l.Info().Time("deadline", deadline).Msg("start to create standby series index")
- standby, err = sic.newNextIdx(ctx)
+ if liveTime <= 0 || liveTime > sic.standbyLiveTime {
+ return nil
+ }
+ return sic.createStandby(ctx, now, deadline)
+}
+
+func (sic *seriesIndexController[T, O]) handleStandby(ctx context.Context, now, deadline time.Time) error {
+ sic.l.Info().Time("deadline", deadline).Msg("start to swap series index")
+
+ if sic.standby == nil {
+ var err error
+ sic.standby, err = sic.newIdx(ctx, now)
if err != nil {
return err
}
- sic.Lock()
- sic.standby = standby
- sic.Unlock()
}
+
+ standby := sic.hot
+ sic.hot = sic.standby
+ sic.standby = nil
+
+ if err := standby.Close(); err != nil {
+ sic.l.Warn().Err(err).Msg("fail to close standby series index")
+ }
+
+ lfs.MustRMAll(standby.path)
+ sic.l.Info().Str("path", standby.path).Msg("dropped series index")
+ lfs.SyncPath(sic.location)
+
+ return nil
+}
+
+func (sic *seriesIndexController[T, O]) createStandby(ctx context.Context, now, deadline time.Time) error {
+ if sic.standby != nil {
+ return nil
+ }
+ sic.l.Info().Time("deadline", deadline).Msg("start to create standby series index")
+ standby, err := sic.newNextIdx(ctx, now)
+ if err != nil {
+ sic.l.Error().Err(err).Msg("fail to create standby series index")
+ return err
+ }
+
+ sic.standby = standby
return nil
}
diff --git a/banyand/internal/storage/rotation.go b/banyand/internal/storage/rotation.go
index 025f1c4d..808d95da 100644
--- a/banyand/internal/storage/rotation.go
+++ b/banyand/internal/storage/rotation.go
@@ -46,11 +46,11 @@ func (d *database[T, O]) startRotationTask() error {
rt := newRetentionTask(d, d.opts.TTL)
go func(rt *retentionTask[T, O]) {
for ts := range d.tsEventCh {
+ logger.Infof("rotation task start at %s", time.Unix(0, ts).Format(time.RFC3339))
func(ts int64) {
d.rotationProcessOn.Store(true)
defer d.rotationProcessOn.Store(false)
t := time.Unix(0, ts)
- d.logger.Info().Time("ts", t).Msg("rotation task triggered")
rt.run(t, d.logger)
shardsRef := d.sLst.Load()
if shardsRef == nil {
@@ -71,7 +71,6 @@ func (d *database[T, O]) startRotationTask() error {
gap := latest.End.UnixNano() - ts
// gap <=0 means the event is from the future
// the segment will be created by a written event directly
- d.logger.Info().Time("ts", t).Time("end", latest.End).Msg("check segment creation")
if gap <= 0 || gap > newSegmentTimeGap {
return
}
@@ -83,7 +82,6 @@ func (d *database[T, O]) startRotationTask() error {
}(s)
}
}(ts)
- d.logger.Info().Time("ts", time.Unix(0, ts)).Msg("rotation task finished")
}
}(rt)
return d.scheduler.Register("retention", rt.option, rt.expr, rt.run)
@@ -92,9 +90,9 @@ func (d *database[T, O]) startRotationTask() error {
type retentionTask[T TSTable, O any] struct {
database *database[T, O]
running chan struct{}
+ expr string
option cron.ParseOption
duration time.Duration
- expr string
}
func newRetentionTask[T TSTable, O any](database *database[T, O], ttl IntervalRule) *retentionTask[T, O] {
@@ -130,7 +128,7 @@ func (rc *retentionTask[T, O]) run(now time.Time, l *logger.Logger) bool {
}
}
stdDeadline := rc.database.opts.TTL.Unit.standard(deadline)
- if err := rc.database.indexController.run(stdDeadline); err != nil {
+ if err := rc.database.indexController.run(now, stdDeadline); err != nil {
l.Error().Err(err)
}
return true
diff --git a/banyand/internal/storage/rotation_test.go b/banyand/internal/storage/rotation_test.go
index 5ecddb1a..869e5f63 100644
--- a/banyand/internal/storage/rotation_test.go
+++ b/banyand/internal/storage/rotation_test.go
@@ -186,8 +186,8 @@ func (m *MockTSTable) Close() error {
return nil
}
-var MockTSTableCreator = func(fileSystem fs.FileSystem, root string, position common.Position,
- l *logger.Logger, timeRange timestamp.TimeRange, option any,
+var MockTSTableCreator = func(_ fs.FileSystem, _ string, _ common.Position,
+ _ *logger.Logger, _ timestamp.TimeRange, _ any,
) (*MockTSTable, error) {
return &MockTSTable{}, nil
}
diff --git a/banyand/internal/storage/segment.go b/banyand/internal/storage/segment.go
index 2e935244..1a69a6c4 100644
--- a/banyand/internal/storage/segment.go
+++ b/banyand/internal/storage/segment.go
@@ -290,16 +290,13 @@ func (sc *segmentController[T, O]) load(start, end time.Time, root string) (seg
}
func (sc *segmentController[T, O]) remove(deadline time.Time) (err error) {
- sc.l.Info().Time("deadline", deadline).Msg("start to remove before deadline")
for _, s := range sc.segments() {
if s.Before(deadline) {
- if e := sc.l.Debug(); e.Enabled() {
- e.Stringer("segment", s).Msg("start to remove data in a segment")
- }
s.delete()
sc.Lock()
sc.removeSeg(s.id)
sc.Unlock()
+ sc.l.Info().Stringer("segment", s).Msg("removed a segment")
}
s.DecRef()
}
diff --git a/banyand/internal/storage/tsdb.go b/banyand/internal/storage/tsdb.go
index 264e3e96..ba53b107 100644
--- a/banyand/internal/storage/tsdb.go
+++ b/banyand/internal/storage/tsdb.go
@@ -68,18 +68,18 @@ func generateSegID(unit IntervalUnit, suffix int) segmentID {
}
type database[T TSTable, O any] struct {
- logger *logger.Logger
- lock fs.File
- indexController *seriesIndexController[T, O]
- scheduler *timestamp.Scheduler
- p common.Position
- location string
- sLst atomic.Pointer[[]*shard[T, O]]
- opts TSDBOpts[T, O]
- tsEventCh chan int64
- latestTickTime atomic.Int64
- rotationProcessOn atomic.Bool
+ lock fs.File
+ logger *logger.Logger
+ indexController *seriesIndexController[T, O]
+ scheduler *timestamp.Scheduler
+ sLst atomic.Pointer[[]*shard[T, O]]
+ tsEventCh chan int64
+ p common.Position
+ location string
+ opts TSDBOpts[T, O]
+ latestTickTime atomic.Int64
sync.RWMutex
+ rotationProcessOn atomic.Bool
}
func (d *database[T, O]) Close() error {
diff --git a/banyand/stream/elements.go b/banyand/stream/elements.go
index 0206671e..b5f104d3 100644
--- a/banyand/stream/elements.go
+++ b/banyand/stream/elements.go
@@ -149,8 +149,8 @@ type elementsInTable struct {
}
type elementsInGroup struct {
- tsdb storage.TSDB[*tsTable, option]
-
- docs index.Documents
- tables []*elementsInTable
+ tsdb storage.TSDB[*tsTable, option]
+ docs index.Documents
+ tables []*elementsInTable
+ latestTS int64
}
diff --git a/banyand/stream/write.go b/banyand/stream/write.go
index bd0b1f31..04e9ea73 100644
--- a/banyand/stream/write.go
+++ b/banyand/stream/write.go
@@ -68,6 +68,9 @@ func (w *writeCallback) handle(dst map[string]*elementsInGroup, writeEvent *stre
}
dst[gn] = eg
}
+ if eg.latestTS < ts {
+ eg.latestTS = ts
+ }
var et *elementsInTable
for i := range eg.tables {
@@ -224,6 +227,7 @@ func (w *writeCallback) Rev(message bus.Message) (resp bus.Message) {
}
for i := range groups {
g := groups[i]
+ g.tsdb.Tick(g.latestTS)
for j := range g.tables {
es := g.tables[j]
es.tsTable.Table().mustAddElements(&es.elements)
diff --git a/pkg/test/flags/flags.go b/pkg/test/flags/flags.go
index 978a4f81..0cbd084f 100644
--- a/pkg/test/flags/flags.go
+++ b/pkg/test/flags/flags.go
@@ -30,6 +30,7 @@ var (
// EventuallyTimeout is the timeout of async time cases execution.
EventuallyTimeout time.Duration
+ // NeverTimeout is the timeout of async time cases execution.
NeverTimeout time.Duration
// LogLevel is the log level of test cases.
diff --git a/pkg/test/helpers/file.go b/pkg/test/helpers/file.go
new file mode 100644
index 00000000..aa1245ad
--- /dev/null
+++ b/pkg/test/helpers/file.go
@@ -0,0 +1,63 @@
+// Licensed to Apache Software Foundation (ASF) under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Apache Software Foundation (ASF) licenses this file to you 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 helpers
+
+import (
+ "fmt"
+ "os"
+ "path/filepath"
+
+ "github.com/dustin/go-humanize"
+)
+
+// PrintDiskUsage prints the disk usage of the specified directory and its subdirectories.
+func PrintDiskUsage(dir string, maxDepth, curDepth int) {
+ // Calculate the total size of all files and directories within the directory
+ var totalSize int64
+ err := filepath.Walk(dir, func(_ string, info os.FileInfo, err error) error {
+ if err != nil {
+ return err
+ }
+ if info.Mode().IsRegular() {
+ totalSize += info.Size()
+ }
+ return nil
+ })
+ if err != nil {
+ fmt.Fprintf(os.Stderr, "Error: %v\n", err)
+ return
+ }
+
+ // Print the disk usage of the current directory
+ fmt.Printf("%s: %s\n", dir, humanize.Bytes(uint64(totalSize)))
+
+ // Recursively print the disk usage of subdirectories
+ if curDepth < maxDepth {
+ files, err := os.ReadDir(dir)
+ if err != nil {
+ fmt.Fprintf(os.Stderr, "Error: %v\n", err)
+ return
+ }
+ for _, file := range files {
+ if file.IsDir() {
+ subdir := filepath.Join(dir, file.Name())
+ PrintDiskUsage(subdir, maxDepth, curDepth+1)
+ }
+ }
+ }
+}
diff --git a/pkg/test/setup/setup.go b/pkg/test/setup/setup.go
index 8fdcbfec..ea9c04ff 100644
--- a/pkg/test/setup/setup.go
+++ b/pkg/test/setup/setup.go
@@ -102,7 +102,7 @@ func standaloneServer(path string, ports []int, schemaLoaders []SchemaLoader, ce
endpoint := fmt.Sprintf("http://%s:%d", host, ports[2])
ff := []string{
"--logging-env=dev",
- "--logging-level=error",
+ "--logging-level=info",
"--grpc-host=" + host,
fmt.Sprintf("--grpc-port=%d", ports[0]),
"--http-host=" + host,
diff --git a/pkg/test/space.go b/pkg/test/space.go
index 14448bee..a288b877 100644
--- a/pkg/test/space.go
+++ b/pkg/test/space.go
@@ -19,7 +19,10 @@ package test
import (
"fmt"
+ "log"
"os"
+ "path/filepath"
+ "strings"
"github.com/stretchr/testify/require"
)
@@ -47,3 +50,21 @@ func NewSpace() (tempDir string, deferFunc func(), err error) {
}
}, err
}
+
+// Cleanup removes all files and directories created by the test.
+func Cleanup() {
+ files, err := os.ReadDir(os.TempDir())
+ if err != nil {
+ log.Fatal(err)
+ return
+ }
+
+ for _, file := range files {
+ if strings.HasPrefix(file.Name(), "banyandb-") {
+ err = os.RemoveAll(filepath.Join(os.TempDir(), file.Name()))
+ if err != nil {
+ log.Fatal(err)
+ }
+ }
+ }
+}
diff --git a/pkg/test/stream/testdata/group.json b/pkg/test/stream/testdata/group.json
index ac044783..b9e8dc80 100644
--- a/pkg/test/stream/testdata/group.json
+++ b/pkg/test/stream/testdata/group.json
@@ -11,7 +11,7 @@
},
"ttl": {
"unit": "UNIT_DAY",
- "num": 7
+ "num": 3
}
},
"updated_at": "2021-04-15T01:30:15.01Z"
diff --git a/pkg/timestamp/range.go b/pkg/timestamp/range.go
index 6bae472a..15a4c0e2 100644
--- a/pkg/timestamp/range.go
+++ b/pkg/timestamp/range.go
@@ -29,6 +29,7 @@ type TimeRange struct {
IncludeEnd bool
}
+// Before returns whether the TimeRange is before the other time.
func (t TimeRange) Before(other time.Time) bool {
if t.IncludeEnd {
return t.End.Before(other)
diff --git a/test/integration/load/load_suite_test.go b/test/integration/load/load_suite_test.go
index 357fcfcc..f2696d4f 100644
--- a/test/integration/load/load_suite_test.go
+++ b/test/integration/load/load_suite_test.go
@@ -19,9 +19,13 @@ package integration_load_test
import (
"context"
+ "os"
+ "path"
+ "strings"
"testing"
"time"
+ "github.com/montanaflynn/stats"
. "github.com/onsi/ginkgo/v2"
. "github.com/onsi/gomega"
"github.com/onsi/gomega/gleak"
@@ -32,6 +36,7 @@ import (
streamv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/stream/v1"
"github.com/apache/skywalking-banyandb/pkg/grpchelper"
"github.com/apache/skywalking-banyandb/pkg/logger"
+ "github.com/apache/skywalking-banyandb/pkg/test"
"github.com/apache/skywalking-banyandb/pkg/test/flags"
"github.com/apache/skywalking-banyandb/pkg/test/helpers"
"github.com/apache/skywalking-banyandb/pkg/test/setup"
@@ -43,79 +48,107 @@ func TestIntegrationLoad(t *testing.T) {
RunSpecs(t, "Integration Load Suite", Label("integration", "slow"))
}
+const (
+ minutes = 10 * 24 * 60
+ interval = 10 * time.Second
+ queryInterval = time.Hour
+ reportInterval = 24 * time.Hour
+)
+
var _ = Describe("Load Test Suit", func() {
var (
connection *grpc.ClientConn
- now time.Time
deferFunc func()
goods []gleak.Goroutine
- addr string
+ dir string
)
+ queryFn := func(now time.Time, dur time.Duration) time.Duration {
+ query := &streamv1.QueryRequest{
+ Name: "sw",
+ Groups: []string{"default"},
+ Projection: &modelv1.TagProjection{
+ TagFamilies: []*modelv1.TagProjection_TagFamily{
+ {
+ Name: "searchable",
+ Tags: []string{"trace_id"},
+ },
+ },
+ },
+ }
+ query.TimeRange = helpers.TimeRange(helpers.Args{Input: "all", Duration: dur}, helpers.SharedContext{
+ Connection: connection,
+ BaseTime: now,
+ })
+ c := streamv1.NewStreamServiceClient(connection)
+ ctx := context.Background()
+ GinkgoWriter.Printf("querying at %s\n", now)
+ start := time.Now()
+ resp, err := c.Query(ctx, query)
+ Expect(err).NotTo(HaveOccurred())
+ latency := time.Since(start)
+ size := len(resp.GetElements())
+ Expect(size).Should(BeNumerically(">", 0))
+ GinkgoWriter.Printf("query result: %s elements using %s \n", size, latency)
+ return latency
+ }
+
BeforeEach(func() {
Expect(logger.Init(logger.Logging{
Env: "dev",
- Level: flags.LogLevel,
+ Level: "debug",
})).Should(Succeed())
+ test.Cleanup()
goods = gleak.Goroutines()
+ var addr string
addr, _, deferFunc = setup.Standalone()
Eventually(
helpers.HealthCheck(addr, 10*time.Second, 10*time.Second, grpc.WithTransportCredentials(insecure.NewCredentials())),
flags.EventuallyTimeout).Should(Succeed())
var err error
- connection, err = grpchelper.Conn(addr, 10*time.Second, grpc.WithTransportCredentials(insecure.NewCredentials()))
+ connection, err = grpchelper.Conn(addr, 10*time.Second,
+ grpc.WithTransportCredentials(insecure.NewCredentials()))
Expect(err).NotTo(HaveOccurred())
- days := 7
- hours := 24
- minutes := 60
- interval := 10 * time.Second
- c := time.Now()
- for i := 0; i < days; i++ {
- date := c.Add(-time.Hour * time.Duration((days-i)*24))
- for h := 0; h < hours; h++ {
- hour := date.Add(time.Hour * time.Duration(h))
- start := time.Now()
- for j := 0; j < minutes; j++ {
- n := hour.Add(time.Minute * time.Duration(j))
- ns := n.UnixNano()
- now = time.Unix(0, ns-ns%int64(time.Minute))
- // stream
- cases_stream_data.Write(connection, "data.json", now, interval)
- }
- logger.Infof("written stream in %s took %s \n", hour, time.Since(start))
+ files, err := os.ReadDir(os.TempDir())
+ Expect(err).NotTo(HaveOccurred())
+
+ for _, file := range files {
+ if strings.HasPrefix(file.Name(), "banyandb-") {
+ dir = path.Join(os.TempDir(), file.Name())
+ break
}
}
- Expect(connection.Close()).To(Succeed())
+ Expect(dir).NotTo(BeEmpty())
})
- It("should read data", func() {
- var err error
- connection, err = grpchelper.Conn(addr, 10*time.Second,
- grpc.WithTransportCredentials(insecure.NewCredentials()))
- Expect(err).NotTo(HaveOccurred())
- sharedContext := helpers.SharedContext{
- Connection: connection,
- BaseTime: now,
- }
- query := &streamv1.QueryRequest{
- Name: "sw",
- Groups: []string{"default"},
- Projection: &modelv1.TagProjection{
- TagFamilies: []*modelv1.TagProjection_TagFamily{
- {
- Name: "searchable",
- Tags: []string{"trace_id"},
- },
- },
- },
+ It("should pass without any error after running 10 days with 1 day segment and 3 days ttl", func() {
+ ns := time.Now().UnixNano()
+ now := time.Unix(0, ns-ns%int64(time.Minute))
+ var lastQueryTime, lastReportTime time.Time
+ latest1HourQueryLatencyData := make([]float64, 0)
+ allQueryLatencyData := make([]float64, 0)
+ for i := 0; i < minutes; i++ {
+ GinkgoWriter.Printf("writing data at %s\n", now)
+ cases_stream_data.Write(connection, "data.json", now, interval)
+ if now.Sub(lastQueryTime) > queryInterval {
+ latency := queryFn(now, time.Hour)
+ latest1HourQueryLatencyData = append(latest1HourQueryLatencyData, float64(latency.Milliseconds()))
+ latency = queryFn(now, 11*24*time.Hour)
+ allQueryLatencyData = append(allQueryLatencyData, float64(latency.Milliseconds()))
+ lastQueryTime = now
+ }
+ if now.Sub(lastReportTime) > reportInterval {
+ logger.Infof("reporting at %s\n", now)
+ // Print files in the directory with prefix "banyandb-" in "tmp" directory
+ helpers.PrintDiskUsage(dir, 4, 0)
+ analysis("latest 1 hour query latency", latest1HourQueryLatencyData)
+ analysis("all query latency", allQueryLatencyData)
+ lastReportTime = now
+ latest1HourQueryLatencyData = latest1HourQueryLatencyData[:0]
+ allQueryLatencyData = allQueryLatencyData[:0]
+ }
+ now = now.Add(time.Minute)
}
- query.TimeRange = helpers.TimeRange(helpers.Args{Input: "all", Duration: 1 * time.Hour}, sharedContext)
- c := streamv1.NewStreamServiceClient(sharedContext.Connection)
- ctx := context.Background()
- resp, err := c.Query(ctx, query)
- Expect(err).NotTo(HaveOccurred())
- GinkgoWriter.Printf("query result: %s elements\n", resp.GetElements())
- Expect(len(resp.GetElements())).To(BeNumerically(">", 0))
})
AfterEach(func() {
@@ -126,3 +159,15 @@ var _ = Describe("Load Test Suit", func() {
Eventually(gleak.Goroutines, flags.EventuallyTimeout).ShouldNot(gleak.HaveLeaked(goods))
})
})
+
+func analysis(name string, data []float64) {
+ min, _ := stats.Min(data)
+ max, _ := stats.Max(data)
+ mean, _ := stats.Mean(data)
+ median, _ := stats.Median(data)
+ p90, _ := stats.Percentile(data, 90)
+ p95, _ := stats.Percentile(data, 95)
+ p98, _ := stats.Percentile(data, 98)
+ p99, _ := stats.Percentile(data, 99)
+ logger.Infof("%s: min: %f, max: %f, mean: %f, median: %f, p90: %f, p95: %f, p98: %f, p99: %f\n", name, min, max, mean, median, p90, p95, p98, p99)
+}
diff --git a/test/stress/istio/istio_suite_test.go b/test/stress/istio/istio_suite_test.go
index 9c94b654..d77078fc 100644
--- a/test/stress/istio/istio_suite_test.go
+++ b/test/stress/istio/istio_suite_test.go
@@ -29,7 +29,6 @@ import (
"testing"
"time"
- "github.com/dustin/go-humanize"
g "github.com/onsi/ginkgo/v2"
"github.com/onsi/gomega"
"github.com/pkg/errors"
@@ -76,7 +75,7 @@ var _ = g.Describe("Istio", func() {
g.DeferCleanup(func() {
time.Sleep(time.Minute)
closerServerFunc()
- printDiskUsage(measurePath, 5, 0)
+ helpers.PrintDiskUsage(measurePath, 5, 0)
deferFn()
})
gomega.Eventually(helpers.HealthCheck(addr, 10*time.Second, 10*time.Second, grpc.WithTransportCredentials(insecure.NewCredentials())),
@@ -225,42 +224,6 @@ func ReadAndWriteFromFile(filePath string, conn *grpc.ClientConn) (int, error) {
return writeCount, flush(false)
}
-func printDiskUsage(dir string, maxDepth, curDepth int) {
- // Calculate the total size of all files and directories within the directory
- var totalSize int64
- err := filepath.Walk(dir, func(_ string, info os.FileInfo, err error) error {
- if err != nil {
- return err
- }
- if info.Mode().IsRegular() {
- totalSize += info.Size()
- }
- return nil
- })
- if err != nil {
- fmt.Fprintf(os.Stderr, "Error: %v\n", err)
- return
- }
-
- // Print the disk usage of the current directory
- fmt.Printf("%s: %s\n", dir, humanize.Bytes(uint64(totalSize)))
-
- // Recursively print the disk usage of subdirectories
- if curDepth < maxDepth {
- files, err := os.ReadDir(dir)
- if err != nil {
- fmt.Fprintf(os.Stderr, "Error: %v\n", err)
- return
- }
- for _, file := range files {
- if file.IsDir() {
- subdir := filepath.Join(dir, file.Name())
- printDiskUsage(subdir, maxDepth, curDepth+1)
- }
- }
- }
-}
-
type clientCounter struct {
bytesSent atomic.Int64
totalLatency atomic.Int64
diff --git a/test/stress/trace/docker-compose-single.yaml b/test/stress/trace/docker-compose-single.yaml
index e990ad3b..78a4e39e 100644
--- a/test/stress/trace/docker-compose-single.yaml
+++ b/test/stress/trace/docker-compose-single.yaml
@@ -57,6 +57,7 @@ services:
image: "docker.io/hanahmily/data-generator:${SW_OAP_COMMIT}"
environment:
SW_STORAGE: banyandb
+ SW_STORAGE_BANYANDB_FLUSH_INTERVAL: 5
ports:
- 12800:12800
volumes:
diff --git a/test/stress/trace/env b/test/stress/trace/env
index 864bcd53..457b0ab2 100644
--- a/test/stress/trace/env
+++ b/test/stress/trace/env
@@ -14,4 +14,4 @@
# limitations under the License.
-SW_OAP_COMMIT=ad80d9cca7a862b719f7dd8229cb9bfe01b77c3c
+SW_OAP_COMMIT=b698e7f13a9c62db9ab652a6cf76aeb745d2c38f
diff --git a/test/stress/trace/segment.tpl.json b/test/stress/trace/segment.tpl.json
index 3597234a..5c5a9fae 100644
--- a/test/stress/trace/segment.tpl.json
+++ b/test/stress/trace/segment.tpl.json
@@ -31,6 +31,11 @@
"min": 1,
"max": 1
},
+ "now": {
+ "type": "time",
+ "stepMillisecond": 300000,
+ "waitMillisecond": 1000
+ },
"tags": {
"type": "randomList",
"size": 5,