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/12/25 11:14:18 UTC
(skywalking-banyandb) branch index-cache updated (0c2566d1 -> 88d84ba8)
This is an automated email from the ASF dual-hosted git repository.
hanahmily pushed a change to branch index-cache
in repository https://gitbox.apache.org/repos/asf/skywalking-banyandb.git
discard 0c2566d1 Add a cache to improve the performance of the series index write
new 88d84ba8 Add a cache to improve the performance of the series index write
This update added new revisions after undoing existing revisions.
That is to say, some revisions that were in the old version of the
branch are not in the new version. This situation occurs
when a user --force pushes a change and generates a repository
containing something like this:
* -- * -- B -- O -- O -- O (0c2566d1)
\
N -- N -- N refs/heads/index-cache (88d84ba8)
You should already have received notification emails for all of the O
revisions, and so the following emails describe only the N revisions
from the common base, B.
Any revisions marked "omit" are not gone; other references still
refer to them. Any revisions marked "discard" are gone forever.
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
dist/LICENSE | 4 +++-
...0-probing.txt => license-github.com-VictoriaMetrics-fastcache.txt} | 2 +-
...its-and-blooms-bitset.txt => license-github.com-golang-snappy.txt} | 2 +-
3 files changed, 5 insertions(+), 3 deletions(-)
copy dist/licenses/{license-github.com-xiang90-probing.txt => license-github.com-VictoriaMetrics-fastcache.txt} (96%)
copy dist/licenses/{license-github.com-bits-and-blooms-bitset.txt => license-github.com-golang-snappy.txt} (95%)
(skywalking-banyandb) 01/01: Add a cache to improve the performance of the series index write
Posted by ha...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
hanahmily pushed a commit to branch index-cache
in repository https://gitbox.apache.org/repos/asf/skywalking-banyandb.git
commit 88d84ba88492f95f6a3103e40d8b66ec3d4abf2e
Author: Gao Hongtao <ha...@gmail.com>
AuthorDate: Wed Dec 25 19:11:19 2024 +0800
Add a cache to improve the performance of the series index write
Signed-off-by: Gao Hongtao <ha...@gmail.com>
---
CHANGES.md | 1 +
banyand/internal/storage/index.go | 11 +++++----
banyand/internal/storage/index_test.go | 4 ++--
banyand/internal/storage/rotation.go | 5 ++++
banyand/internal/storage/segment.go | 2 +-
banyand/internal/storage/tsdb.go | 1 +
banyand/measure/measure.go | 6 +++--
banyand/measure/metadata.go | 3 ++-
banyand/measure/service.go | 4 +++-
banyand/stream/metadata.go | 1 +
banyand/stream/service.go | 2 ++
banyand/stream/stream.go | 2 ++
dist/LICENSE | 4 +++-
...icense-github.com-VictoriaMetrics-fastcache.txt | 22 +++++++++++++++++
dist/licenses/license-github.com-golang-snappy.txt | 27 +++++++++++++++++++++
go.mod | 4 +++-
go.sum | 12 ++++++++--
pkg/index/index.go | 1 +
pkg/index/inverted/inverted.go | 14 +++++++----
pkg/index/inverted/metrics.go | 28 ++++++++++++++++++++++
20 files changed, 135 insertions(+), 19 deletions(-)
diff --git a/CHANGES.md b/CHANGES.md
index 6742fc3b..c5a29241 100644
--- a/CHANGES.md
+++ b/CHANGES.md
@@ -14,6 +14,7 @@ Release Notes.
- TopN: Group top n pre-calculation result by the group key in the new introduced `_top_n_result` measure, which is used to store the pre-calculation result.
- Index Mode: Index `measure_nam` and `tags` in `entity` to improve the query performance.
- Encoding: Improve the performance of encoding and decoding the variable-length int64.
+- Index: Add a cache to improve the performance of the series index write.
### Bug Fixes
diff --git a/banyand/internal/storage/index.go b/banyand/internal/storage/index.go
index fc6eea59..1e34de74 100644
--- a/banyand/internal/storage/index.go
+++ b/banyand/internal/storage/index.go
@@ -50,15 +50,18 @@ type seriesIndex struct {
p common.Position
}
-func newSeriesIndex(ctx context.Context, root string, flushTimeoutSeconds int64, metrics *inverted.Metrics) (*seriesIndex, error) {
+func newSeriesIndex(ctx context.Context, root string, flushTimeoutSeconds int64, cacheMaxBytes int,
+ metrics *inverted.Metrics,
+) (*seriesIndex, error) {
si := &seriesIndex{
l: logger.Fetch(ctx, "series_index"),
p: common.GetPosition(ctx),
}
opts := inverted.StoreOpts{
- Path: path.Join(root, "sidx"),
- Logger: si.l,
- BatchWaitSec: flushTimeoutSeconds,
+ Path: path.Join(root, "sidx"),
+ Logger: si.l,
+ BatchWaitSec: flushTimeoutSeconds,
+ CacheMaxBytes: cacheMaxBytes,
}
if metrics != nil {
opts.Metrics = metrics
diff --git a/banyand/internal/storage/index_test.go b/banyand/internal/storage/index_test.go
index d02c70d4..143e7ee5 100644
--- a/banyand/internal/storage/index_test.go
+++ b/banyand/internal/storage/index_test.go
@@ -37,7 +37,7 @@ import (
func TestSeriesIndex_Primary(t *testing.T) {
ctx := context.Background()
path, fn := setUp(require.New(t))
- si, err := newSeriesIndex(ctx, path, 0, nil)
+ si, err := newSeriesIndex(ctx, path, 0, 0, nil)
require.NoError(t, err)
defer func() {
require.NoError(t, si.Close())
@@ -67,7 +67,7 @@ func TestSeriesIndex_Primary(t *testing.T) {
require.NoError(t, si.Insert(docs))
// Restart the index
require.NoError(t, si.Close())
- si, err = newSeriesIndex(ctx, path, 0, nil)
+ si, err = newSeriesIndex(ctx, path, 0, 0, nil)
require.NoError(t, err)
tests := []struct {
name string
diff --git a/banyand/internal/storage/rotation.go b/banyand/internal/storage/rotation.go
index 2066ef90..81f242e7 100644
--- a/banyand/internal/storage/rotation.go
+++ b/banyand/internal/storage/rotation.go
@@ -62,6 +62,11 @@ func (d *database[T, O]) startRotationTask() error {
ss[i].DecRef()
}
}()
+ for i := range ss {
+ if ss[i].End.UnixNano() < ts {
+ ss[i].index.store.Reset()
+ }
+ }
latest := ss[len(ss)-1]
gap := latest.End.UnixNano() - ts
// gap <=0 means the event is from the future
diff --git a/banyand/internal/storage/segment.go b/banyand/internal/storage/segment.go
index 0451c649..cb7047e6 100644
--- a/banyand/internal/storage/segment.go
+++ b/banyand/internal/storage/segment.go
@@ -72,7 +72,7 @@ func (sc *segmentController[T, O]) openSegment(ctx context.Context, startTime, e
})
options := sc.getOptions()
id := generateSegID(options.SegmentInterval.Unit, suffixInteger)
- sir, err := newSeriesIndex(ctx, path, options.SeriesIndexFlushTimeoutSeconds, sc.indexMetrics)
+ sir, err := newSeriesIndex(ctx, path, options.SeriesIndexFlushTimeoutSeconds, options.SeriesIndexCacheMaxBytes, sc.indexMetrics)
if err != nil {
return nil, errors.Wrap(errOpenDatabase, errors.WithMessage(err, "create series index controller failed").Error())
}
diff --git a/banyand/internal/storage/tsdb.go b/banyand/internal/storage/tsdb.go
index 5d62bc82..167f0380 100644
--- a/banyand/internal/storage/tsdb.go
+++ b/banyand/internal/storage/tsdb.go
@@ -61,6 +61,7 @@ type TSDBOpts[T TSTable, O any] struct {
TTL IntervalRule
SeriesIndexFlushTimeoutSeconds int64
ShardNum uint32
+ SeriesIndexCacheMaxBytes int
}
type (
diff --git a/banyand/measure/measure.go b/banyand/measure/measure.go
index 4c4f85bd..b2b7f286 100644
--- a/banyand/measure/measure.go
+++ b/banyand/measure/measure.go
@@ -29,6 +29,7 @@ import (
"github.com/apache/skywalking-banyandb/pkg/logger"
"github.com/apache/skywalking-banyandb/pkg/partition"
"github.com/apache/skywalking-banyandb/pkg/query/logical"
+ "github.com/apache/skywalking-banyandb/pkg/run"
"github.com/apache/skywalking-banyandb/pkg/schema"
"github.com/apache/skywalking-banyandb/pkg/timestamp"
)
@@ -45,8 +46,9 @@ const (
)
type option struct {
- mergePolicy *mergePolicy
- flushTimeout time.Duration
+ mergePolicy *mergePolicy
+ flushTimeout time.Duration
+ seriesCacheMaxSize run.Bytes
}
type measure struct {
diff --git a/banyand/measure/metadata.go b/banyand/measure/metadata.go
index 07de2db2..882a83f8 100644
--- a/banyand/measure/metadata.go
+++ b/banyand/measure/metadata.go
@@ -265,8 +265,8 @@ type supplier struct {
pipeline queue.Queue
omr observability.MetricsRegistry
l *logger.Logger
- option option
path string
+ option option
}
func newSupplier(path string, svc *service) *supplier {
@@ -311,6 +311,7 @@ func (s *supplier) OpenDB(groupSchema *commonv1.Group) (resourceSchema.DB, error
TTL: storage.MustToIntervalRule(groupSchema.ResourceOpts.Ttl),
Option: s.option,
SeriesIndexFlushTimeoutSeconds: s.option.flushTimeout.Nanoseconds() / int64(time.Second),
+ SeriesIndexCacheMaxBytes: int(s.option.seriesCacheMaxSize),
StorageMetricsFactory: factory,
}
return storage.OpenTSDB(
diff --git a/banyand/measure/service.go b/banyand/measure/service.go
index 22a58918..bd7aed21 100644
--- a/banyand/measure/service.go
+++ b/banyand/measure/service.go
@@ -60,8 +60,8 @@ type service struct {
omr observability.MetricsRegistry
schemaRepo *schemaRepo
l *logger.Logger
- option option
root string
+ option option
}
func (s *service) Measure(metadata *commonv1.Metadata) (Measure, error) {
@@ -82,6 +82,8 @@ func (s *service) FlagSet() *run.FlagSet {
flagS.DurationVar(&s.option.flushTimeout, "measure-flush-timeout", defaultFlushTimeout, "the memory data timeout of measure")
s.option.mergePolicy = newDefaultMergePolicy()
flagS.VarP(&s.option.mergePolicy.maxFanOutSize, "measure-max-fan-out-size", "", "the upper bound of a single file size after merge of measure")
+ s.option.seriesCacheMaxSize = run.Bytes(10 << 20)
+ flagS.VarP(&s.option.seriesCacheMaxSize, "measure-series-cache-max-size", "", "the max size of series cache in each group")
return flagS
}
diff --git a/banyand/stream/metadata.go b/banyand/stream/metadata.go
index 0c6a1c43..bde097d8 100644
--- a/banyand/stream/metadata.go
+++ b/banyand/stream/metadata.go
@@ -295,6 +295,7 @@ func (s *supplier) OpenDB(groupSchema *commonv1.Group) (resourceSchema.DB, error
TTL: storage.MustToIntervalRule(groupSchema.ResourceOpts.Ttl),
Option: s.option,
SeriesIndexFlushTimeoutSeconds: s.option.flushTimeout.Nanoseconds() / int64(time.Second),
+ SeriesIndexCacheMaxBytes: int(s.option.seriesCacheMaxSize),
StorageMetricsFactory: s.omr.With(storageScope.ConstLabels(meter.ToLabelPairs(common.DBLabelNames(), p.DBLabelValues()))),
}
return storage.OpenTSDB(
diff --git a/banyand/stream/service.go b/banyand/stream/service.go
index ae54e82a..fcf5c945 100644
--- a/banyand/stream/service.go
+++ b/banyand/stream/service.go
@@ -82,6 +82,8 @@ func (s *service) FlagSet() *run.FlagSet {
flagS.DurationVar(&s.option.elementIndexFlushTimeout, "element-index-flush-timeout", defaultFlushTimeout, "the elementIndex timeout of stream")
s.option.mergePolicy = newDefaultMergePolicy()
flagS.VarP(&s.option.mergePolicy.maxFanOutSize, "stream-max-fan-out-size", "", "the upper bound of a single file size after merge of stream")
+ s.option.seriesCacheMaxSize = run.Bytes(10 << 20)
+ flagS.VarP(&s.option.seriesCacheMaxSize, "measure-series-cache-max-size", "", "the max size of series cache in each group")
return flagS
}
diff --git a/banyand/stream/stream.go b/banyand/stream/stream.go
index 1243a38e..406e9bc4 100644
--- a/banyand/stream/stream.go
+++ b/banyand/stream/stream.go
@@ -29,6 +29,7 @@ import (
"github.com/apache/skywalking-banyandb/pkg/logger"
"github.com/apache/skywalking-banyandb/pkg/partition"
"github.com/apache/skywalking-banyandb/pkg/query/model"
+ "github.com/apache/skywalking-banyandb/pkg/run"
"github.com/apache/skywalking-banyandb/pkg/schema"
)
@@ -46,6 +47,7 @@ type option struct {
mergePolicy *mergePolicy
flushTimeout time.Duration
elementIndexFlushTimeout time.Duration
+ seriesCacheMaxSize run.Bytes
}
// Query allow to retrieve elements in a series of streams.
diff --git a/dist/LICENSE b/dist/LICENSE
index be91e422..081e0c5b 100644
--- a/dist/LICENSE
+++ b/dist/LICENSE
@@ -178,7 +178,7 @@
Apache-2.0 licenses
========================================================================
- github.com/SkyAPM/bluge v0.0.0-20241204041252-279547fc72c6 Apache-2.0
+ github.com/SkyAPM/bluge v0.0.0-20241225104157-e54f64be56e8 Apache-2.0
github.com/SkyAPM/ice v0.0.0-20241108011032-c3d8eea75118 Apache-2.0
github.com/apache/skywalking-cli v0.0.0-20240227151024-ee371a210afe Apache-2.0
github.com/blevesearch/segment v0.9.1 Apache-2.0
@@ -268,6 +268,7 @@ BSD-3-Clause licenses
github.com/fsnotify/fsnotify v1.8.0 BSD-3-Clause
github.com/gogo/protobuf v1.3.2 BSD-3-Clause
github.com/golang/protobuf v1.5.4 BSD-3-Clause
+ github.com/golang/snappy v0.0.4 BSD-3-Clause
github.com/google/go-cmp v0.6.0 BSD-3-Clause
github.com/google/uuid v1.6.0 BSD-3-Clause
github.com/grpc-ecosystem/grpc-gateway v1.16.0 BSD-3-Clause
@@ -314,6 +315,7 @@ MIT licenses
========================================================================
github.com/SkyAPM/clock v1.3.1-0.20220809233656-dc7607c94a97 MIT
+ github.com/VictoriaMetrics/fastcache v1.12.2 MIT
github.com/axiomhq/hyperloglog v0.2.2 MIT
github.com/beorn7/perks v1.0.1 MIT
github.com/blevesearch/go-porterstemmer v1.0.3 MIT
diff --git a/dist/licenses/license-github.com-VictoriaMetrics-fastcache.txt b/dist/licenses/license-github.com-VictoriaMetrics-fastcache.txt
new file mode 100644
index 00000000..9a8145e5
--- /dev/null
+++ b/dist/licenses/license-github.com-VictoriaMetrics-fastcache.txt
@@ -0,0 +1,22 @@
+The MIT License (MIT)
+
+Copyright (c) 2018 VictoriaMetrics
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.
+
diff --git a/dist/licenses/license-github.com-golang-snappy.txt b/dist/licenses/license-github.com-golang-snappy.txt
new file mode 100644
index 00000000..6050c10f
--- /dev/null
+++ b/dist/licenses/license-github.com-golang-snappy.txt
@@ -0,0 +1,27 @@
+Copyright (c) 2011 The Snappy-Go Authors. All rights reserved.
+
+Redistribution and use in source and binary forms, with or without
+modification, are permitted provided that the following conditions are
+met:
+
+ * Redistributions of source code must retain the above copyright
+notice, this list of conditions and the following disclaimer.
+ * Redistributions in binary form must reproduce the above
+copyright notice, this list of conditions and the following disclaimer
+in the documentation and/or other materials provided with the
+distribution.
+ * Neither the name of Google Inc. nor the names of its
+contributors may be used to endorse or promote products derived from
+this software without specific prior written permission.
+
+THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+"AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
diff --git a/go.mod b/go.mod
index a6ab4b33..b2a0be6e 100644
--- a/go.mod
+++ b/go.mod
@@ -49,8 +49,10 @@ require (
)
require (
+ github.com/VictoriaMetrics/fastcache v1.12.2 // indirect
github.com/cpuguy83/go-md2man/v2 v2.0.6 // indirect
github.com/go-task/slim-sprig/v3 v3.0.0 // indirect
+ github.com/golang/snappy v0.0.4 // indirect
github.com/kamstrup/intmap v0.5.1 // indirect
github.com/machinebox/graphql v0.2.2 // indirect
github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect
@@ -159,7 +161,7 @@ require (
replace (
github.com/benbjohnson/clock v1.3.0 => github.com/SkyAPM/clock v1.3.1-0.20220809233656-dc7607c94a97
- github.com/blugelabs/bluge => github.com/SkyAPM/bluge v0.0.0-20241204041252-279547fc72c6
+ github.com/blugelabs/bluge => github.com/SkyAPM/bluge v0.0.0-20241225104157-e54f64be56e8
github.com/blugelabs/bluge_segment_api => github.com/zinclabs/bluge_segment_api v1.0.0
github.com/blugelabs/ice => github.com/SkyAPM/ice v0.0.0-20241108011032-c3d8eea75118
)
diff --git a/go.sum b/go.sum
index d6cbed3e..4e1a293a 100644
--- a/go.sum
+++ b/go.sum
@@ -6,12 +6,16 @@ github.com/OneOfOne/xxhash v1.2.2/go.mod h1:HSdplMjZKSmBqAxg5vPj2TmRDmfkzw+cTzAE
github.com/RoaringBitmap/roaring v0.9.4/go.mod h1:icnadbWcNyfEHlYdr+tDlOTih1Bf/h+rzPpv4sbomAA=
github.com/RoaringBitmap/roaring v1.9.4 h1:yhEIoH4YezLYT04s1nHehNO64EKFTop/wBhxv2QzDdQ=
github.com/RoaringBitmap/roaring v1.9.4/go.mod h1:6AXUsoIEzDTFFQCe1RbGA6uFONMhvejWj5rqITANK90=
-github.com/SkyAPM/bluge v0.0.0-20241204041252-279547fc72c6 h1:Zwv1x7CUnX1tBqfTmBzM2tAOROMi4ZHJtdbkuBEpHfo=
-github.com/SkyAPM/bluge v0.0.0-20241204041252-279547fc72c6/go.mod h1:6o9wC3xO3qb5Q7VmD1x0r54qQBDpO9+ghGAQvuOHsCU=
+github.com/SkyAPM/bluge v0.0.0-20241225104157-e54f64be56e8 h1:bWJ8yNeaph7oe4S/kvoAM0dNLNamfsOzxaajhLvTAks=
+github.com/SkyAPM/bluge v0.0.0-20241225104157-e54f64be56e8/go.mod h1:rriyHHsTidJ4UYFiVDIZyXCFJxQYs5FGpZmIfuvSqPA=
github.com/SkyAPM/clock v1.3.1-0.20220809233656-dc7607c94a97 h1:FKuhJ+6n/DHspGeLleeNbziWnKr9gHKYN4q7NcoCp4s=
github.com/SkyAPM/clock v1.3.1-0.20220809233656-dc7607c94a97/go.mod h1:2xGRl9H1pllhxTbEGO1W3gDkip8P9GQaHPni/wpdR44=
github.com/SkyAPM/ice v0.0.0-20241108011032-c3d8eea75118 h1:Ja62sgOCp2qPTd8Xmldv1U83v11IRIsh6KlB7UaFLj4=
github.com/SkyAPM/ice v0.0.0-20241108011032-c3d8eea75118/go.mod h1:DoQeb0Ee86LyruZSL77Ddscfk/THJ38x453CRCnGEPI=
+github.com/VictoriaMetrics/fastcache v1.12.2 h1:N0y9ASrJ0F6h0QaC3o6uJb3NIZ9VKLjCM7NQbSmF7WI=
+github.com/VictoriaMetrics/fastcache v1.12.2/go.mod h1:AmC+Nzz1+3G2eCPapF6UcsnkThDcMsQicp4xDukwJYI=
+github.com/allegro/bigcache v1.2.1-0.20190218064605-e24eb225f156 h1:eMwmnE/GDgah4HI848JfFxHt+iPb26b4zyfspmqY0/8=
+github.com/allegro/bigcache v1.2.1-0.20190218064605-e24eb225f156/go.mod h1:Cb/ax3seSYIx7SuZdm2G2xzfwmv3TPSk2ucNfQESPXM=
github.com/antihax/optional v1.0.0/go.mod h1:uupD/76wgC+ih3iEmQUL+0Ugr19nfwCT1kdvxnR2qWY=
github.com/apache/skywalking-cli v0.0.0-20240227151024-ee371a210afe h1:zIc2yfpc/vMpfTtWprCVpca6CMJwb6X9cknqAoFeEFo=
github.com/apache/skywalking-cli v0.0.0-20240227151024-ee371a210afe/go.mod h1:pu6Q19Xs38FSfy/IwnJGAMilO+W58/ugM8aMfLzw+i0=
@@ -41,6 +45,7 @@ github.com/cenkalti/backoff/v4 v4.3.0/go.mod h1:Y3VNntkOUPxTVeUxJ/G5vcM//AlwfmyY
github.com/census-instrumentation/opencensus-proto v0.2.1/go.mod h1:f6KPmirojxKA12rnyqOA5BBL4O983OfeGPqjHWSTneU=
github.com/cespare/xxhash v1.1.0 h1:a6HrQnmkObjyL+Gs60czilIUGqrzKutQD6XZog3p+ko=
github.com/cespare/xxhash v1.1.0/go.mod h1:XrSqR1VqqWfGrhpAt58auRo0WTKS1nRRg3ghfAqPWnc=
+github.com/cespare/xxhash/v2 v2.2.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs=
github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
github.com/client9/misspell v0.3.4/go.mod h1:qj6jICC3Q7zFZvVWo7KLAzC3yx5G7kyvSDkc90ppPyw=
@@ -122,6 +127,8 @@ github.com/golang/protobuf v1.5.0/go.mod h1:FsONVRAS9T7sI+LIUmWTfcYkHO4aIWwzhcaS
github.com/golang/protobuf v1.5.2/go.mod h1:XVQd3VNwM+JqD3oG2Ue2ip4fOMUkwXdXDdiuN0vRsmY=
github.com/golang/protobuf v1.5.4 h1:i7eJL8qZTpSEXOPTxNKhASYpMn+8e5Q6AdndVa1dWek=
github.com/golang/protobuf v1.5.4/go.mod h1:lnTiLA8Wa4RWRcIUkrtSVa5nRhsEGBg48fD6rSs7xps=
+github.com/golang/snappy v0.0.4 h1:yAGX7huGHXlcLOEtBnF4w7FQwA26wojNCwOYAEhLjQM=
+github.com/golang/snappy v0.0.4/go.mod h1:/XxbfmMg8lxefKM7IXC3fBNl/7bRcc72aCRzEWrmP2Q=
github.com/google/btree v1.1.3 h1:CVpQJjYgC4VbzxeGVHfvZrv1ctoYCAI8vbl07Fcxlyg=
github.com/google/btree v1.1.3/go.mod h1:qOPhT0dTNdNzV6Z/lhRX0YXUafgPLFUh+gZMl761Gm4=
github.com/google/go-cmp v0.2.0/go.mod h1:oXzfMopK8JAjlY9xF4vHSVASa0yLyX7SntLO5aqRK0M=
@@ -443,6 +450,7 @@ golang.org/x/sys v0.5.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
golang.org/x/sys v0.8.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
golang.org/x/sys v0.12.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.14.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
golang.org/x/sys v0.16.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
golang.org/x/sys v0.28.0 h1:Fksou7UEQUWlKvIdsqzJmUmCX3cZuD2+P3XyyzwMhlA=
golang.org/x/sys v0.28.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
diff --git a/pkg/index/index.go b/pkg/index/index.go
index 6538558b..c9b0042a 100644
--- a/pkg/index/index.go
+++ b/pkg/index/index.go
@@ -345,6 +345,7 @@ type Store interface {
Writer
Searcher
CollectMetrics(...string)
+ Reset()
}
// Series represents a series in an index.
diff --git a/pkg/index/inverted/inverted.go b/pkg/index/inverted/inverted.go
index 55255fc5..a54b4f19 100644
--- a/pkg/index/inverted/inverted.go
+++ b/pkg/index/inverted/inverted.go
@@ -74,10 +74,11 @@ var _ index.Store = (*store)(nil)
// StoreOpts wraps options to create an inverted index repository.
type StoreOpts struct {
- Logger *logger.Logger
- Metrics *Metrics
- Path string
- BatchWaitSec int64
+ Logger *logger.Logger
+ Metrics *Metrics
+ Path string
+ BatchWaitSec int64
+ CacheMaxBytes int
}
type store struct {
@@ -154,6 +155,7 @@ func NewStore(opts StoreOpts) (index.SeriesStore, error) {
indexConfig = indexConfig.WithUnsafeBatches().
WithPersisterNapTimeMSec(int(opts.BatchWaitSec * 1000))
}
+ indexConfig.CacheMaxBytes = opts.CacheMaxBytes
config := bluge.DefaultConfigWithIndexConfig(indexConfig)
config.DefaultSearchAnalyzer = Analyzers[index.AnalyzerKeyword]
config.Logger = log.New(opts.Logger, opts.Logger.Module(), 0)
@@ -176,6 +178,10 @@ func (s *store) Close() error {
return s.writer.Close()
}
+func (s *store) Reset() {
+ s.writer.ResetCache()
+}
+
func (s *store) Iterator(ctx context.Context, fieldKey index.FieldKey, termRange index.RangeOpts, order modelv1.Sort,
preLoadSize int,
) (iter index.FieldIterator[*index.DocumentResult], err error) {
diff --git a/pkg/index/inverted/metrics.go b/pkg/index/inverted/metrics.go
index 98b6d7a3..cf2cbf27 100644
--- a/pkg/index/inverted/metrics.go
+++ b/pkg/index/inverted/metrics.go
@@ -46,6 +46,13 @@ type Metrics struct {
curOnDiskFiles meter.Gauge
totalDocCount meter.Gauge
+
+ cacheGetCalls meter.Gauge
+ cacheSetCalls meter.Gauge
+ cacheMisses meter.Gauge
+ cacheEntriesCount meter.Gauge
+ cacheBytesSize meter.Gauge
+ cacheMaxBytesSize meter.Gauge
}
// NewMetrics creates a new metrics for the inverted index.
@@ -73,6 +80,13 @@ func NewMetrics(factory *observability.Factory, labelNames ...string) *Metrics {
curOnDiskFiles: factory.NewGauge("inverted_index_cur_on_disk_files", labelNames...),
totalDocCount: factory.NewGauge("inverted_index_total_doc_count", labelNames...),
+
+ cacheGetCalls: factory.NewGauge("inverted_index_cache_get_calls", labelNames...),
+ cacheSetCalls: factory.NewGauge("inverted_index_cache_set_calls", labelNames...),
+ cacheMisses: factory.NewGauge("inverted_index_cache_misses", labelNames...),
+ cacheEntriesCount: factory.NewGauge("inverted_index_cache_entries_count", labelNames...),
+ cacheBytesSize: factory.NewGauge("inverted_index_cache_bytes_size", labelNames...),
+ cacheMaxBytesSize: factory.NewGauge("inverted_index_cache_max_bytes_size", labelNames...),
}
}
@@ -106,6 +120,13 @@ func (m *Metrics) DeleteAll(labelValues ...string) {
m.totalFileSegments.Delete(labelValues...)
m.curOnDiskBytes.Delete(labelValues...)
m.curOnDiskFiles.Delete(labelValues...)
+
+ m.cacheGetCalls.Delete(labelValues...)
+ m.cacheSetCalls.Delete(labelValues...)
+ m.cacheMisses.Delete(labelValues...)
+ m.cacheEntriesCount.Delete(labelValues...)
+ m.cacheBytesSize.Delete(labelValues...)
+ m.cacheMaxBytesSize.Delete(labelValues...)
}
func (s *store) CollectMetrics(labelValues ...string) {
@@ -140,6 +161,13 @@ func (s *store) CollectMetrics(labelValues ...string) {
s.metrics.curOnDiskBytes.Set(float64(status.CurOnDiskBytes), labelValues...)
s.metrics.curOnDiskFiles.Set(float64(status.CurOnDiskFiles), labelValues...)
+ s.metrics.cacheGetCalls.Set(float64(status.CacheGetCalls), labelValues...)
+ s.metrics.cacheSetCalls.Set(float64(status.CacheSetCalls), labelValues...)
+ s.metrics.cacheMisses.Set(float64(status.CacheMisses), labelValues...)
+ s.metrics.cacheEntriesCount.Set(float64(status.CacheEntriesCount), labelValues...)
+ s.metrics.cacheBytesSize.Set(float64(status.CacheBytesSize), labelValues...)
+ s.metrics.cacheMaxBytesSize.Set(float64(status.CacheMaxBytesSize), labelValues...)
+
r, err := s.writer.Reader()
if err != nil {
return