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 2025/01/18 00:59:09 UTC
(skywalking-banyandb) 02/02: Benchmark the query
This is an automated email from the ASF dual-hosted git repository.
hanahmily pushed a commit to branch stream
in repository https://gitbox.apache.org/repos/asf/skywalking-banyandb.git
commit bf34bb700f4176fa810f9c854bae13d9d6823d16
Author: Gao Hongtao <ha...@gmail.com>
AuthorDate: Sat Jan 18 00:52:32 2025 +0000
Benchmark the query
Signed-off-by: Gao Hongtao <ha...@gmail.com>
---
CHANGES.md | 1 +
pkg/test/measure/etcd.go | 3 +
pkg/test/stream/etcd.go | 7 +
test/integration/load/load_suite_test.go | 206 ++++++++++++++++++---------
test/stress/istio/testdata/groups/group.yaml | 5 +-
5 files changed, 150 insertions(+), 72 deletions(-)
diff --git a/CHANGES.md b/CHANGES.md
index f02912f5..163a5a84 100644
--- a/CHANGES.md
+++ b/CHANGES.md
@@ -20,6 +20,7 @@ Release Notes.
- Add the max disk usage threshold for the `Measure`, `Stream`, and `Property` to control the disk usage.
- Add the "api version" service to gRPC and HTTP server.
- Metadata: Wait for the existing registration to be removed before registering the node.
+- Stream: Introduce the batch scan to improve the performance of the query and limit the memory usage.
### Bug Fixes
diff --git a/pkg/test/measure/etcd.go b/pkg/test/measure/etcd.go
index e647cf9f..4838c2eb 100644
--- a/pkg/test/measure/etcd.go
+++ b/pkg/test/measure/etcd.go
@@ -89,6 +89,9 @@ func loadSchema[T proto.Message](dir string, resource T, loadFn func(resource T)
return err
}
if err := loadFn(resource); err != nil {
+ if errors.Is(err, schema.ErrGRPCAlreadyExists) {
+ return nil
+ }
return err
}
}
diff --git a/pkg/test/stream/etcd.go b/pkg/test/stream/etcd.go
index 56ad5766..93dcf0fb 100644
--- a/pkg/test/stream/etcd.go
+++ b/pkg/test/stream/etcd.go
@@ -21,6 +21,7 @@ package stream
import (
"context"
"embed"
+ "errors"
"path"
"google.golang.org/protobuf/encoding/protojson"
@@ -28,6 +29,7 @@ import (
commonv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/common/v1"
databasev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/database/v1"
"github.com/apache/skywalking-banyandb/banyand/metadata/schema"
+ "github.com/apache/skywalking-banyandb/pkg/logger"
)
const (
@@ -56,6 +58,11 @@ func PreloadSchema(ctx context.Context, e schema.Registry) error {
if err := protojson.Unmarshal([]byte(groupJSON), g); err != nil {
return err
}
+ _, err := e.GetGroup(ctx, g.Metadata.Name)
+ if !errors.Is(err, schema.ErrGRPCResourceNotFound) {
+ logger.Infof("group %s already exists", g.Metadata.Name)
+ return nil
+ }
if innerErr := e.CreateGroup(ctx, g); innerErr != nil {
return innerErr
}
diff --git a/test/integration/load/load_suite_test.go b/test/integration/load/load_suite_test.go
index 2a39e6a3..c2ac8402 100644
--- a/test/integration/load/load_suite_test.go
+++ b/test/integration/load/load_suite_test.go
@@ -20,8 +20,8 @@ package integration_load_test
import (
"context"
"os"
- "path"
- "strings"
+ "path/filepath"
+ "sync"
"testing"
"time"
@@ -29,11 +29,13 @@ import (
. "github.com/onsi/ginkgo/v2"
. "github.com/onsi/gomega"
"github.com/onsi/gomega/gleak"
+ "github.com/onsi/gomega/gmeasure"
"google.golang.org/grpc"
"google.golang.org/grpc/credentials/insecure"
modelv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/model/v1"
streamv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/stream/v1"
+ "github.com/apache/skywalking-banyandb/pkg/cgroups"
"github.com/apache/skywalking-banyandb/pkg/grpchelper"
"github.com/apache/skywalking-banyandb/pkg/logger"
"github.com/apache/skywalking-banyandb/pkg/pool"
@@ -42,6 +44,7 @@ import (
"github.com/apache/skywalking-banyandb/pkg/test/gmatcher"
"github.com/apache/skywalking-banyandb/pkg/test/helpers"
"github.com/apache/skywalking-banyandb/pkg/test/setup"
+ "github.com/apache/skywalking-banyandb/pkg/timestamp"
cases_stream_data "github.com/apache/skywalking-banyandb/test/cases/stream/data"
)
@@ -65,92 +68,117 @@ var _ = Describe("Load Test Suit", func() {
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() {
+ JustBeforeEach(func() {
Expect(logger.Init(logger.Logging{
Env: "dev",
Level: "debug",
})).Should(Succeed())
- test.Cleanup()
goods = gleak.Goroutines()
+ ports, err := test.AllocateFreePorts(4)
+ Expect(err).NotTo(HaveOccurred())
var addr string
- addr, _, deferFunc = setup.Standalone()
+ addr, _, deferFunc = setup.ClosableStandalone(dir, ports)
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()))
Expect(err).NotTo(HaveOccurred())
- 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
+ When("running 10 days with 1 day segment and 3 days ttl", func() {
+ BeforeEach(func() {
+ test.Cleanup()
+ var err error
+ dir, err = os.MkdirTemp("", "banyandb-test-*")
+ Expect(err).NotTo(HaveOccurred())
+ })
+
+ It("should pass", 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, "sw", now, interval)
+ if now.Sub(lastQueryTime) > queryInterval {
+ latency := queryFn(now, time.Hour, connection)
+ latest1HourQueryLatencyData = append(latest1HourQueryLatencyData, float64(latency.Milliseconds()))
+ latency = queryFn(now, 11*24*time.Hour, connection)
+ 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)
}
- }
- Expect(dir).NotTo(BeEmpty())
+ })
})
- 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, "sw", 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
+ When("running 3 days with 1 day segment and 3 days ttl", func() {
+ loadData := false
+ BeforeEach(func() {
+ var err error
+ dir = filepath.Join(os.TempDir(), "banyandb-load-query-test")
+ err = os.MkdirAll(dir, 0o755)
+ Expect(err).NotTo(HaveOccurred())
+ info, err := os.Stat(dir)
+ Expect(err).NotTo(HaveOccurred())
+ Expect(info.IsDir()).Should(BeTrue())
+ entries, err := os.ReadDir(dir)
+ Expect(err).NotTo(HaveOccurred())
+ if len(entries) > 0 {
+ loadData = true
}
- 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]
+ logger.Infof("load data: %v\n", loadData)
+ })
+
+ It("should pass the benchmark", Label("benchmark"), func() {
+ if !loadData {
+ ns := time.Now().UnixNano()
+ now := time.Unix(0, ns-ns%int64(time.Minute))
+ duration := 3 * 24 * 60
+ workerSize := cgroups.CPUs() / 2
+ logger.Infof("worker size: %d\n", workerSize)
+ var wg sync.WaitGroup
+ wg.Add(workerSize)
+ for w := 0; w < workerSize; w++ {
+ go func(offset int) {
+ for i := offset; i < duration; i += workerSize {
+ n := now.Add(time.Duration(i) * time.Minute)
+ cases_stream_data.Write(connection, "sw", n, interval)
+ time.Sleep(100 * time.Millisecond)
+ if (i-offset)%100 == 0 {
+ logger.Infof("worker %d writing data at %s\n", w, n)
+ }
+ }
+ wg.Done()
+ }(w)
+ }
+ wg.Wait()
}
- now = now.Add(time.Minute)
- }
+ tr := timestamp.DefaultTimeRange
+ start := tr.Begin.AsTime()
+ dur := tr.End.AsTime().Sub(start)
+
+ experiment := gmeasure.NewExperiment("Scanning all data")
+ AddReportEntry(experiment.Name, experiment)
+ experiment.Sample(func(idx int) {
+ experiment.MeasureDuration("scanning", func() {
+ queryPureFn(start, dur, connection)
+ })
+ }, gmeasure.SamplingConfig{N: 100, Duration: time.Minute})
+ })
})
AfterEach(func() {
@@ -174,3 +202,41 @@ func analysis(name string, data []float64) {
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, minVal, maxVal, mean, median, p90, p95, p98, p99)
}
+
+func queryFn(now time.Time, dur time.Duration, connection *grpc.ClientConn) time.Duration {
+ GinkgoWriter.Printf("querying at %s\n", now)
+ start := time.Now()
+ size := queryPureFn(now, dur, connection)
+ latency := time.Since(start)
+ Expect(size).Should(BeNumerically(">", 0))
+ GinkgoWriter.Printf("query result: %s elements using %s \n", size, latency)
+ return latency
+}
+
+func queryPureFn(now time.Time, dur time.Duration, connection *grpc.ClientConn) int {
+ query := &streamv1.QueryRequest{
+ Name: "sw",
+ Groups: []string{"default"},
+ Projection: &modelv1.TagProjection{
+ TagFamilies: []*modelv1.TagProjection_TagFamily{
+ {
+ Name: "searchable",
+ Tags: []string{"trace_id", "state", "service_id", "service_instance_id", "duration", "start_time"},
+ },
+ {
+ Name: "data",
+ Tags: []string{"data_binary"},
+ },
+ },
+ },
+ }
+ query.TimeRange = helpers.TimeRange(helpers.Args{Input: "all", Duration: dur}, helpers.SharedContext{
+ Connection: connection,
+ BaseTime: now,
+ })
+ c := streamv1.NewStreamServiceClient(connection)
+ ctx := context.Background()
+ resp, err := c.Query(ctx, query)
+ Expect(err).NotTo(HaveOccurred())
+ return len(resp.GetElements())
+}
diff --git a/test/stress/istio/testdata/groups/group.yaml b/test/stress/istio/testdata/groups/group.yaml
index 1031c1a2..712b66c6 100644
--- a/test/stress/istio/testdata/groups/group.yaml
+++ b/test/stress/istio/testdata/groups/group.yaml
@@ -110,13 +110,14 @@
num: 3
unit: UNIT_DAY
updatedAt: null
-- catalog: CATALOG_UNSPECIFIED
+- catalog: CATALOG_PROPERTY
metadata:
createRevision: "0"
group: ""
id: 0
modRevision: "0"
name: sw
- resourceOpts: null
+ resourceOpts:
+ shardNum: 1
updatedAt: null