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/10 00:01:18 UTC
(skywalking-banyandb) 01/01: Fix lint issues
This is an automated email from the ASF dual-hosted git repository.
hanahmily pushed a commit to branch max-disk
in repository https://gitbox.apache.org/repos/asf/skywalking-banyandb.git
commit bb28c6166a5cf3daca1b80cd3c7a13c95a47d033
Author: Gao Hongtao <ha...@gmail.com>
AuthorDate: Thu Jan 9 22:14:27 2025 +0800
Fix lint issues
Signed-off-by: Gao Hongtao <ha...@gmail.com>
---
CHANGES.md | 1 +
api/common/id.go | 10 ++---
api/data/stream.go | 2 +-
api/proto/banyandb/cluster/v1/rpc.proto | 11 ++++++
api/proto/banyandb/measure/v1/write.proto | 2 +-
api/proto/banyandb/stream/v1/write.proto | 2 +-
banyand/dquery/measure.go | 2 +-
banyand/dquery/stream.go | 2 +-
banyand/dquery/topn.go | 2 +-
banyand/liaison/grpc/measure.go | 11 ++++--
banyand/liaison/grpc/property.go | 2 +-
banyand/liaison/grpc/stream.go | 9 ++++-
banyand/measure/metadata_test.go | 2 +-
banyand/measure/service.go | 6 +++
banyand/measure/write.go | 5 ++-
banyand/property/listener.go | 20 ++++++++--
banyand/property/service.go | 28 ++++++++-----
banyand/query/processor.go | 4 +-
banyand/query/processor_topn.go | 2 +-
banyand/queue/local.go | 13 ++++---
banyand/queue/pub/batch.go | 23 ++++++-----
banyand/queue/pub/client.go | 47 ++++++++++++++--------
banyand/queue/pub/client_test.go | 3 +-
banyand/queue/pub/pub_suite_test.go | 18 ++++-----
banyand/queue/pub/pub_test.go | 6 ++-
banyand/queue/queue.go | 2 +-
banyand/queue/sub/health.go | 54 ++++++++++----------------
banyand/queue/sub/server.go | 5 ++-
banyand/queue/sub/sub.go | 9 +++--
banyand/stream/metadata_test.go | 2 +-
banyand/stream/service.go | 10 ++++-
banyand/stream/write.go | 7 +++-
docs/api-reference.md | 39 ++++++++++++++++++-
docs/operation/troubleshooting/overhead.md | 4 ++
pkg/bus/bus.go | 5 ++-
pkg/bus/bus_test.go | 6 ++-
test/integration/standalone/other/disk_test.go | 10 +++--
37 files changed, 252 insertions(+), 134 deletions(-)
diff --git a/CHANGES.md b/CHANGES.md
index 4c332011..863bb217 100644
--- a/CHANGES.md
+++ b/CHANGES.md
@@ -17,6 +17,7 @@ Release Notes.
- Index: Add a cache to improve the performance of the series index write.
- Read cpu quota and limit from the cgroup file system to set gomaxprocs.
- Property: Add native storage layer for property.
+- Add the max disk usage threshold for the `Measure`, `Stream`, and `Property` to control the disk usage.
### Bug Fixes
diff --git a/api/common/id.go b/api/common/id.go
index 3c29cda9..668df583 100644
--- a/api/common/id.go
+++ b/api/common/id.go
@@ -121,13 +121,13 @@ type Error struct {
}
// NewError returns a new Error.
-func NewError(tpl string, args ...any) Error {
- return Error{status: modelv1.Status_STATUS_INTERNAL_ERROR, msg: fmt.Sprintf(tpl, args...)}
+func NewError(tpl string, args ...any) *Error {
+ return &Error{status: modelv1.Status_STATUS_INTERNAL_ERROR, msg: fmt.Sprintf(tpl, args...)}
}
// NewErrorWithStatus returns a new Error with status.
-func NewErrorWithStatus(status modelv1.Status, msg string) Error {
- return Error{status: status, msg: msg}
+func NewErrorWithStatus(status modelv1.Status, msg string) *Error {
+ return &Error{status: status, msg: msg}
}
// Status returns the status.
@@ -137,7 +137,7 @@ func (e Error) Status() modelv1.Status {
// Error returns the error msg.
func (e Error) Error() string {
- return e.msg
+ return fmt.Sprintf("code: %s, msg: %s", modelv1.Status_name[int32(e.status)], e.msg)
}
// Node contains the node id and address.
diff --git a/api/data/stream.go b/api/data/stream.go
index e083e7db..7677731b 100644
--- a/api/data/stream.go
+++ b/api/data/stream.go
@@ -29,7 +29,7 @@ var StreamWriteKindVersion = common.KindVersion{
}
// TopicStreamWrite is the stream write topic.
-var TopicStreamWrite = bus.UniTopic(StreamWriteKindVersion.String())
+var TopicStreamWrite = bus.BiTopic(StreamWriteKindVersion.String())
// StreamQueryKindVersion is the version tag of stream query kind.
var StreamQueryKindVersion = common.KindVersion{
diff --git a/api/proto/banyandb/cluster/v1/rpc.proto b/api/proto/banyandb/cluster/v1/rpc.proto
index e2c0d8f1..fc0ea731 100644
--- a/api/proto/banyandb/cluster/v1/rpc.proto
+++ b/api/proto/banyandb/cluster/v1/rpc.proto
@@ -39,6 +39,17 @@ message SendResponse {
string topic = 5;
}
+message HealthCheckRequest {
+ string service_name = 1;
+}
+
+message HealthCheckResponse {
+ string service_name = 1;
+ model.v1.Status status = 2;
+ string error = 3;
+}
+
service Service {
rpc Send(stream SendRequest) returns (stream SendResponse);
+ rpc HealthCheck(HealthCheckRequest) returns (HealthCheckResponse);
}
diff --git a/api/proto/banyandb/measure/v1/write.proto b/api/proto/banyandb/measure/v1/write.proto
index 35493efe..2ac2699c 100644
--- a/api/proto/banyandb/measure/v1/write.proto
+++ b/api/proto/banyandb/measure/v1/write.proto
@@ -65,7 +65,7 @@ message WriteResponse {
// the message_id from request.
uint64 message_id = 1 [(validate.rules).uint64.gt = 0];
// status indicates the request processing result
- model.v1.Status status = 2 [(validate.rules).enum.defined_only = true];
+ string status = 2;
// the metadata from request when request fails
common.v1.Metadata metadata = 3;
}
diff --git a/api/proto/banyandb/stream/v1/write.proto b/api/proto/banyandb/stream/v1/write.proto
index 01b8f221..1f88a61d 100644
--- a/api/proto/banyandb/stream/v1/write.proto
+++ b/api/proto/banyandb/stream/v1/write.proto
@@ -52,7 +52,7 @@ message WriteResponse {
// the message_id from request.
uint64 message_id = 1 [(validate.rules).uint64.gt = 0];
// status indicates the request processing result
- model.v1.Status status = 2 [(validate.rules).enum.defined_only = true];
+ string status = 2;
// the metadata from request when request fails
common.v1.Metadata metadata = 3;
}
diff --git a/banyand/dquery/measure.go b/banyand/dquery/measure.go
index ae79f15d..a1e3ecce 100644
--- a/banyand/dquery/measure.go
+++ b/banyand/dquery/measure.go
@@ -94,7 +94,7 @@ func (p *measureQueryProcessor) Rev(ctx context.Context, message bus.Message) (r
switch d := data.(type) {
case *measurev1.QueryResponse:
d.Trace = tracer.ToProto()
- case common.Error:
+ case *common.Error:
span.Error(errors.New(d.Error()))
resp = bus.NewMessage(bus.MessageID(now), &measurev1.QueryResponse{Trace: tracer.ToProto()})
default:
diff --git a/banyand/dquery/stream.go b/banyand/dquery/stream.go
index 32e9012f..80a31c9d 100644
--- a/banyand/dquery/stream.go
+++ b/banyand/dquery/stream.go
@@ -90,7 +90,7 @@ func (p *streamQueryProcessor) Rev(ctx context.Context, message bus.Message) (re
switch d := data.(type) {
case *streamv1.QueryResponse:
d.Trace = tracer.ToProto()
- case common.Error:
+ case *common.Error:
span.Error(errors.New(d.Error()))
resp = bus.NewMessage(bus.MessageID(now), &streamv1.QueryResponse{Trace: tracer.ToProto()})
default:
diff --git a/banyand/dquery/topn.go b/banyand/dquery/topn.go
index 5641a2e9..067d7f0e 100644
--- a/banyand/dquery/topn.go
+++ b/banyand/dquery/topn.go
@@ -74,7 +74,7 @@ func (t *topNQueryProcessor) Rev(ctx context.Context, message bus.Message) (resp
switch d := data.(type) {
case *measurev1.TopNResponse:
d.Trace = tracer.ToProto()
- case common.Error:
+ case *common.Error:
span.Error(errors.New(d.Error()))
resp = bus.NewMessage(now, &measurev1.TopNResponse{Trace: tracer.ToProto()})
default:
diff --git a/banyand/liaison/grpc/measure.go b/banyand/liaison/grpc/measure.go
index f91ffb6f..7dd38590 100644
--- a/banyand/liaison/grpc/measure.go
+++ b/banyand/liaison/grpc/measure.go
@@ -70,7 +70,7 @@ func (ms *measureService) Write(measure measurev1.MeasureService_WriteServer) er
ms.metrics.totalStreamMsgReceivedErr.Inc(1, metadata.Group, "measure", "write")
}
ms.metrics.totalStreamMsgSent.Inc(1, metadata.Group, "measure", "write")
- if errResp := measure.Send(&measurev1.WriteResponse{Metadata: metadata, Status: status, MessageId: messageId}); errResp != nil {
+ if errResp := measure.Send(&measurev1.WriteResponse{Metadata: metadata, Status: status.String(), MessageId: messageId}); errResp != nil {
logger.Debug().Err(errResp).Msg("failed to send measure write response")
ms.metrics.totalStreamMsgSentErr.Inc(1, metadata.Group, "measure", "write")
}
@@ -165,6 +165,11 @@ func (ms *measureService) Write(measure measurev1.MeasureService_WriteServer) er
_, errWritePub := publisher.Publish(ctx, data.TopicMeasureWrite, message)
if errWritePub != nil {
ms.sampled.Error().Err(errWritePub).RawJSON("written", logger.Proto(writeRequest)).Str("nodeID", nodeID).Msg("failed to send a message")
+ var ce *common.Error
+ if errors.As(errWritePub, &ce) {
+ reply(writeRequest.GetMetadata(), ce.Status(), writeRequest.GetMessageId(), measure, ms.sampled)
+ continue
+ }
reply(writeRequest.GetMetadata(), modelv1.Status_STATUS_INTERNAL_ERROR, writeRequest.GetMessageId(), measure, ms.sampled)
continue
}
@@ -225,7 +230,7 @@ func (ms *measureService) Query(ctx context.Context, req *measurev1.QueryRequest
switch d := data.(type) {
case *measurev1.QueryResponse:
return d, nil
- case common.Error:
+ case *common.Error:
return nil, errors.WithMessage(errQueryMsg, d.Error())
}
return nil, nil
@@ -263,7 +268,7 @@ func (ms *measureService) TopN(ctx context.Context, topNRequest *measurev1.TopNR
switch d := data.(type) {
case *measurev1.TopNResponse:
return d, nil
- case common.Error:
+ case *common.Error:
return nil, errors.WithMessage(errQueryMsg, d.Error())
}
return nil, nil
diff --git a/banyand/liaison/grpc/property.go b/banyand/liaison/grpc/property.go
index b64267d8..721bdb83 100644
--- a/banyand/liaison/grpc/property.go
+++ b/banyand/liaison/grpc/property.go
@@ -333,7 +333,7 @@ func (ps *propertyServer) Query(ctx context.Context, req *propertyv1.QueryReques
if span != nil {
span.AddSubTrace(v.Trace)
}
- case common.Error:
+ case *common.Error:
err = multierr.Append(err, errors.New(v.Error()))
}
}
diff --git a/banyand/liaison/grpc/stream.go b/banyand/liaison/grpc/stream.go
index 886e5a5f..c7c0c245 100644
--- a/banyand/liaison/grpc/stream.go
+++ b/banyand/liaison/grpc/stream.go
@@ -70,7 +70,7 @@ func (s *streamService) Write(stream streamv1.StreamService_WriteServer) error {
s.metrics.totalStreamMsgReceivedErr.Inc(1, metadata.Group, "stream", "write")
}
s.metrics.totalStreamMsgReceived.Inc(1, metadata.Group, "stream", "write")
- if errResp := stream.Send(&streamv1.WriteResponse{Metadata: metadata, Status: status, MessageId: messageId}); errResp != nil {
+ if errResp := stream.Send(&streamv1.WriteResponse{Metadata: metadata, Status: status.String(), MessageId: messageId}); errResp != nil {
logger.Debug().Err(errResp).Msg("failed to send stream write response")
s.metrics.totalStreamMsgSentErr.Inc(1, metadata.Group, "stream", "write")
}
@@ -158,6 +158,11 @@ func (s *streamService) Write(stream streamv1.StreamService_WriteServer) error {
message := bus.NewBatchMessageWithNode(bus.MessageID(time.Now().UnixNano()), nodeID, iwr)
_, errWritePub := publisher.Publish(ctx, data.TopicStreamWrite, message)
if errWritePub != nil {
+ var ce *common.Error
+ if errors.As(errWritePub, &ce) {
+ reply(writeEntity.GetMetadata(), ce.Status(), writeEntity.GetMessageId(), stream, s.sampled)
+ continue
+ }
s.sampled.Error().Err(errWritePub).RawJSON("written", logger.Proto(writeEntity)).Str("nodeID", nodeID).Msg("failed to send a message")
reply(writeEntity.GetMetadata(), modelv1.Status_STATUS_INTERNAL_ERROR, writeEntity.GetMessageId(), stream, s.sampled)
continue
@@ -224,7 +229,7 @@ func (s *streamService) Query(ctx context.Context, req *streamv1.QueryRequest) (
switch d := data.(type) {
case *streamv1.QueryResponse:
return d, nil
- case common.Error:
+ case *common.Error:
return nil, errors.WithMessage(errQueryMsg, d.Error())
}
return nil, nil
diff --git a/banyand/measure/metadata_test.go b/banyand/measure/metadata_test.go
index 015c07f0..733266cf 100644
--- a/banyand/measure/metadata_test.go
+++ b/banyand/measure/metadata_test.go
@@ -358,7 +358,7 @@ func queryAllMeasurements(svcs *services, expectedSize int, newTag []string, new
}
resp = d
return true
- case common.Error:
+ case *common.Error:
Fail(d.Error())
default:
Fail("unexpected data type")
diff --git a/banyand/measure/service.go b/banyand/measure/service.go
index 10a74f68..e5eb353c 100644
--- a/banyand/measure/service.go
+++ b/banyand/measure/service.go
@@ -93,6 +93,12 @@ func (s *service) Validate() error {
if s.root == "" {
return errEmptyRootPath
}
+ if s.maxDiskUsagePercent < 0 {
+ return errors.New("measure-max-disk-usage-percen must be greater than or equal to 0")
+ }
+ if s.maxDiskUsagePercent > 100 {
+ return errors.New("measure-max-disk-usage-percen must be less than or equal to 100")
+ }
return nil
}
diff --git a/banyand/measure/write.go b/banyand/measure/write.go
index f22a48f3..da6f8f9d 100644
--- a/banyand/measure/write.go
+++ b/banyand/measure/write.go
@@ -48,6 +48,9 @@ type writeCallback struct {
}
func setUpWriteCallback(l *logger.Logger, schemaRepo *schemaRepo, maxDiskUsagePercent int) bus.MessageListener {
+ if maxDiskUsagePercent > 100 {
+ maxDiskUsagePercent = 100
+ }
return &writeCallback{
l: l,
schemaRepo: schemaRepo,
@@ -55,7 +58,7 @@ func setUpWriteCallback(l *logger.Logger, schemaRepo *schemaRepo, maxDiskUsagePe
}
}
-func (w *writeCallback) CheckHealth() error {
+func (w *writeCallback) CheckHealth() *common.Error {
if w.maxDiskUsagePercent < 1 {
return common.NewErrorWithStatus(modelv1.Status_STATUS_DISK_FULL, "measure is readonly because \"measure-max-disk-usage-percent\" is 0")
}
diff --git a/banyand/property/listener.go b/banyand/property/listener.go
index 5fe6634d..0433b826 100644
--- a/banyand/property/listener.go
+++ b/banyand/property/listener.go
@@ -25,7 +25,9 @@ import (
"google.golang.org/protobuf/proto"
"github.com/apache/skywalking-banyandb/api/common"
+ modelv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/model/v1"
propertyv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/property/v1"
+ "github.com/apache/skywalking-banyandb/banyand/observability"
"github.com/apache/skywalking-banyandb/pkg/bus"
"github.com/apache/skywalking-banyandb/pkg/logger"
"github.com/apache/skywalking-banyandb/pkg/query"
@@ -38,12 +40,22 @@ var (
)
type updateListener struct {
- s *service
- path string
+ s *service
+ l *logger.Logger
+ path string
+ maxDiskUsagePercent int
}
-func (h *updateListener) CheckHealth() error {
- return nil
+func (h *updateListener) CheckHealth() *common.Error {
+ if h.maxDiskUsagePercent < 1 {
+ return common.NewErrorWithStatus(modelv1.Status_STATUS_DISK_FULL, "property is readonly because \"property-max-disk-usage-percent\" is 0")
+ }
+ diskPercent := observability.GetPathUsedPercent(h.path)
+ if diskPercent < h.maxDiskUsagePercent {
+ return nil
+ }
+ h.l.Warn().Int("maxPercent", h.maxDiskUsagePercent).Int("diskPercent", diskPercent).Msg("disk usage is too high, stop writing")
+ return common.NewErrorWithStatus(modelv1.Status_STATUS_DISK_FULL, "disk usage is too high, stop writing")
}
func (h *updateListener) Rev(ctx context.Context, message bus.Message) (resp bus.Message) {
diff --git a/banyand/property/service.go b/banyand/property/service.go
index 620ef380..e3af4fe0 100644
--- a/banyand/property/service.go
+++ b/banyand/property/service.go
@@ -43,21 +43,23 @@ var (
)
type service struct {
- metadata metadata.Repo
- pipeline queue.Server
- omr observability.MetricsRegistry
- l *logger.Logger
- db *database
- close chan struct{}
- root string
- nodeID string
- flushTimeout time.Duration
+ metadata metadata.Repo
+ pipeline queue.Server
+ omr observability.MetricsRegistry
+ l *logger.Logger
+ db *database
+ close chan struct{}
+ root string
+ nodeID string
+ flushTimeout time.Duration
+ maxDiskUsagePercent int
}
func (s *service) FlagSet() *run.FlagSet {
flagS := run.NewFlagSet("storage")
flagS.StringVar(&s.root, "property-root-path", "/tmp", "the root path of database")
flagS.DurationVar(&s.flushTimeout, "property-flush-timeout", defaultFlushTimeout, "the memory data timeout of measure")
+ flagS.IntVar(&s.maxDiskUsagePercent, "property-max-disk-usage-percent", 95, "the maximum disk usage percentage allowed")
return flagS
}
@@ -65,6 +67,12 @@ func (s *service) Validate() error {
if s.root == "" {
return errEmptyRootPath
}
+ if s.maxDiskUsagePercent < 0 {
+ return errors.New("property-max-disk-usage-percent must be greater than or equal to 0")
+ }
+ if s.maxDiskUsagePercent > 100 {
+ return errors.New("property-max-disk-usage-percent must be less than or equal to 100")
+ }
return nil
}
@@ -93,7 +101,7 @@ func (s *service) PreRun(ctx context.Context) error {
return err
}
return multierr.Combine(
- s.pipeline.Subscribe(data.TopicPropertyUpdate, &updateListener{s: s, path: path}),
+ s.pipeline.Subscribe(data.TopicPropertyUpdate, &updateListener{s: s, path: path, maxDiskUsagePercent: s.maxDiskUsagePercent}),
s.pipeline.Subscribe(data.TopicPropertyDelete, &deleteListener{s: s}),
s.pipeline.Subscribe(data.TopicPropertyQuery, &queryListener{s: s}),
)
diff --git a/banyand/query/processor.go b/banyand/query/processor.go
index 7d1ec858..e1b6864a 100644
--- a/banyand/query/processor.go
+++ b/banyand/query/processor.go
@@ -112,7 +112,7 @@ func (p *streamQueryProcessor) Rev(ctx context.Context, message bus.Message) (re
switch d := data.(type) {
case *streamv1.QueryResponse:
d.Trace = tracer.ToProto()
- case common.Error:
+ case *common.Error:
span.Error(errors.New(d.Error()))
resp = bus.NewMessage(bus.MessageID(now), &measurev1.QueryResponse{Trace: tracer.ToProto()})
default:
@@ -202,7 +202,7 @@ func (p *measureQueryProcessor) Rev(ctx context.Context, message bus.Message) (r
switch d := data.(type) {
case *measurev1.QueryResponse:
d.Trace = tracer.ToProto()
- case common.Error:
+ case *common.Error:
span.Error(errors.New(d.Error()))
resp = bus.NewMessage(bus.MessageID(now), &measurev1.QueryResponse{Trace: tracer.ToProto()})
default:
diff --git a/banyand/query/processor_topn.go b/banyand/query/processor_topn.go
index 324661cd..0bb22e7b 100644
--- a/banyand/query/processor_topn.go
+++ b/banyand/query/processor_topn.go
@@ -121,7 +121,7 @@ func (t *topNQueryProcessor) Rev(ctx context.Context, message bus.Message) (resp
switch d := data.(type) {
case *measurev1.TopNResponse:
d.Trace = tracer.ToProto()
- case common.Error:
+ case *common.Error:
span.Error(errors.New(d.Error()))
resp = bus.NewMessage(bus.MessageID(now), &measurev1.QueryResponse{Trace: tracer.ToProto()})
default:
diff --git a/banyand/queue/local.go b/banyand/queue/local.go
index 4c952d33..20bbeb4a 100644
--- a/banyand/queue/local.go
+++ b/banyand/queue/local.go
@@ -20,6 +20,7 @@ package queue
import (
context "context"
+ "errors"
"time"
"github.com/apache/skywalking-banyandb/api/common"
@@ -114,7 +115,7 @@ func (l *localBatchPublisher) Publish(ctx context.Context, topic bus.Topic, mess
return nil, nil
}
-func (l *localBatchPublisher) Close() (map[string]common.Error, error) {
+func (l *localBatchPublisher) Close() (map[string]*common.Error, error) {
if l.local == nil || len(l.messages) == 0 {
return nil, nil
}
@@ -123,9 +124,9 @@ func (l *localBatchPublisher) Close() (map[string]common.Error, error) {
l.messages = nil
l.topic = nil
if err != nil {
- // nolint: errorlint
- if ce, ok := err.(common.Error); ok {
- return map[string]common.Error{"local": ce}, nil
+ var ce *common.Error
+ if errors.As(err, &ce) {
+ return map[string]*common.Error{"local": ce}, nil
}
return nil, err
}
@@ -137,8 +138,8 @@ func (l *localBatchPublisher) Close() (map[string]common.Error, error) {
return nil, err
}
if m.Data() != nil {
- if d, ok := m.Data().(common.Error); ok {
- return map[string]common.Error{"local": d}, nil
+ if d, ok := m.Data().(*common.Error); ok {
+ return map[string]*common.Error{"local": d}, nil
}
}
return nil, nil
diff --git a/banyand/queue/pub/batch.go b/banyand/queue/pub/batch.go
index 17bd6cd2..0c685f3b 100644
--- a/banyand/queue/pub/batch.go
+++ b/banyand/queue/pub/batch.go
@@ -43,7 +43,7 @@ type batchPublisher struct {
pub *pub
streams map[string]writeStream
topic *bus.Topic
- failedNodes map[string]struct{}
+ failedNodes map[string]*common.Error
f batchFuture
timeout time.Duration
}
@@ -102,8 +102,10 @@ func (bp *batchPublisher) Publish(ctx context.Context, topic bus.Topic, messages
return nil, ctx.Err()
default:
}
- if bp.failedNodes != nil && bp.failedNodes[node] == struct{}{} {
- err = multierr.Append(err, fmt.Errorf("node %s is not writable", node))
+ if bp.failedNodes != nil {
+ if ce := bp.failedNodes[node]; ce != nil {
+ err = multierr.Append(err, ce)
+ }
continue
}
var client *client
@@ -117,14 +119,15 @@ func (bp *batchPublisher) Publish(ctx context.Context, topic bus.Topic, messages
err = multierr.Append(err, fmt.Errorf("failed to get client for node %s", node))
return true
}
- if bp.pub.checkWritable(node, topic) {
+ succeed, ce := bp.pub.checkWritable(node, topic)
+ if succeed {
return false
}
if bp.failedNodes == nil {
- bp.failedNodes = make(map[string]struct{})
+ bp.failedNodes = make(map[string]*common.Error)
}
- bp.failedNodes[node] = struct{}{}
- err = multierr.Append(err, fmt.Errorf("node %s is not writable", node))
+ bp.failedNodes[node] = ce
+ err = multierr.Append(err, ce)
return true
}() {
continue
@@ -176,7 +179,7 @@ func (bp *batchPublisher) Publish(ctx context.Context, topic bus.Topic, messages
return nil, err
}
-func (bp *batchPublisher) Close() (cee map[string]common.Error, err error) {
+func (bp *batchPublisher) Close() (cee map[string]*common.Error, err error) {
for i := range bp.streams {
err = multierr.Append(err, bp.streams[i].client.CloseSend())
}
@@ -199,7 +202,7 @@ func (bp *batchPublisher) Close() (cee map[string]common.Error, err error) {
}
}()
}
- cee = make(map[string]common.Error, len(batchEvents))
+ cee = make(map[string]*common.Error, len(batchEvents))
for n, be := range batchEvents {
cee[n] = be.e
}
@@ -209,7 +212,7 @@ func (bp *batchPublisher) Close() (cee map[string]common.Error, err error) {
type batchEvent struct {
n string
t string
- e common.Error
+ e *common.Error
}
type batchFuture struct {
diff --git a/banyand/queue/pub/client.go b/banyand/queue/pub/client.go
index 9ccc224f..cf818cc5 100644
--- a/banyand/queue/pub/client.go
+++ b/banyand/queue/pub/client.go
@@ -283,20 +283,16 @@ func (p *pub) checkClientHealthAndReconnect(conn *grpc.ClientConn, md schema.Met
}
func (p *pub) healthCheck(node string, conn *grpc.ClientConn) bool {
- return p.checkServiceHealth(node, "", conn)
-}
-
-func (p *pub) checkServiceHealth(node string, svc string, conn *grpc.ClientConn) bool {
var resp *grpc_health_v1.HealthCheckResponse
if err := grpchelper.Request(context.Background(), rpcTimeout, func(rpcCtx context.Context) (err error) {
resp, err = grpc_health_v1.NewHealthClient(conn).Check(rpcCtx,
&grpc_health_v1.HealthCheckRequest{
- Service: svc,
+ Service: "",
})
return err
}); err != nil {
if e := p.log.Debug(); e.Enabled() {
- e.Err(err).Str("node", node).Str("svc", svc).Msg("service unhealthy")
+ e.Err(err).Str("node", node).Msg("service unhealthy")
}
return false
}
@@ -306,11 +302,27 @@ func (p *pub) checkServiceHealth(node string, svc string, conn *grpc.ClientConn)
return false
}
-func (p *pub) failover(node string, ce common.Error, topic bus.Topic) {
+func (p *pub) checkServiceHealth(svc string, conn *grpc.ClientConn) *common.Error {
+ client := clusterv1.NewServiceClient(conn)
+ ctx, cancel := context.WithTimeout(context.Background(), rpcTimeout)
+ defer cancel()
+ resp, err := client.HealthCheck(ctx, &clusterv1.HealthCheckRequest{
+ ServiceName: svc,
+ })
+ if err != nil {
+ return common.NewErrorWithStatus(modelv1.Status_STATUS_INTERNAL_ERROR, err.Error())
+ }
+ if resp.Status == modelv1.Status_STATUS_SUCCEED {
+ return nil
+ }
+ return common.NewErrorWithStatus(resp.Status, resp.Error)
+}
+
+func (p *pub) failover(node string, ce *common.Error, topic bus.Topic) {
p.mu.Lock()
defer p.mu.Unlock()
if ce.Status() != modelv1.Status_STATUS_INTERNAL_ERROR {
- _ = p.checkWritable(node, topic)
+ _, _ = p.checkWritable(node, topic)
return
}
if en, evictable := p.evictable[node]; evictable {
@@ -330,21 +342,22 @@ func (p *pub) failover(node string, ce common.Error, topic bus.Topic) {
}
}
-func (p *pub) checkWritable(n string, topic bus.Topic) bool {
+func (p *pub) checkWritable(n string, topic bus.Topic) (bool, *common.Error) {
h, ok := p.handlers[topic]
if !ok {
- return false
+ return false, nil
}
node, ok := p.active[n]
if !ok {
- return false
+ return false, nil
}
- if p.checkServiceHealth(n, topic.String(), node.conn) {
- return true
+ err := p.checkServiceHealth(topic.String(), node.conn)
+ if err == nil {
+ return true, nil
}
h.OnDelete(node.md)
if !p.closer.AddRunning() {
- return false
+ return false, err
}
go func() {
defer p.closer.Done()
@@ -352,7 +365,7 @@ func (p *pub) checkWritable(n string, topic bus.Topic) bool {
for {
select {
case <-time.After(backoff):
- if p.checkServiceHealth(n, topic.String(), node.conn) {
+ if errInternal := p.checkServiceHealth(topic.String(), node.conn); errInternal != nil {
func() {
p.mu.Lock()
defer p.mu.Unlock()
@@ -364,7 +377,7 @@ func (p *pub) checkWritable(n string, topic bus.Topic) bool {
}()
return
}
- p.log.Warn().Str("topic", topic.String()).Str("node", n).Dur("backoff", backoff).Msg("data node can not ingest data")
+ p.log.Warn().Str("topic", topic.String()).Err(err).Str("node", n).Dur("backoff", backoff).Msg("data node can not ingest data")
case <-p.closer.CloseNotify():
return
}
@@ -375,7 +388,7 @@ func (p *pub) checkWritable(n string, topic bus.Topic) bool {
}
}
}()
- return false
+ return false, err
}
func (p *pub) deleteClient(md schema.Metadata) {
diff --git a/banyand/queue/pub/client_test.go b/banyand/queue/pub/client_test.go
index 76fbf286..12678b6e 100644
--- a/banyand/queue/pub/client_test.go
+++ b/banyand/queue/pub/client_test.go
@@ -131,7 +131,8 @@ var _ = ginkgo.Describe("publish clients register/unregister", func() {
_, err := bp.Publish(ctx, data.TopicStreamWrite,
bus.NewBatchMessageWithNode(bus.MessageID(i), "node1", &streamv1.InternalWriteRequest{}),
)
- gomega.Expect(err).Should(gomega.MatchError("node node1 is not writable"))
+ gomega.Expect(err).Should(gomega.MatchError(common.NewErrorWithStatus(
+ modelv1.Status_STATUS_DISK_FULL, modelv1.Status_name[int32(modelv1.Status_STATUS_DISK_FULL)]).Error()))
}
cee, err := bp.Close()
gomega.Expect(err).ShouldNot(gomega.HaveOccurred())
diff --git a/banyand/queue/pub/pub_suite_test.go b/banyand/queue/pub/pub_suite_test.go
index c545b1a6..0d91ffb6 100644
--- a/banyand/queue/pub/pub_suite_test.go
+++ b/banyand/queue/pub/pub_suite_test.go
@@ -18,6 +18,7 @@
package pub
import (
+ "context"
"errors"
"io"
"net"
@@ -134,13 +135,19 @@ func (s *mockServer) Send(stream clusterv1.Service_SendServer) (err error) {
}
}
+func (s *mockServer) HealthCheck(context.Context, *clusterv1.HealthCheckRequest) (*clusterv1.HealthCheckResponse, error) {
+ return &clusterv1.HealthCheckResponse{
+ Status: s.statusCode,
+ Error: s.errMsg,
+ }, nil
+}
+
func setup(address string, code codes.Code, latency time.Duration) func() {
s := grpc.NewServer()
hs := health.NewServer()
- hs.SetServingStatus(data.TopicStreamWrite.String(), grpc_health_v1.HealthCheckResponse_SERVING)
- hs.SetServingStatus(data.TopicMeasureWrite.String(), grpc_health_v1.HealthCheckResponse_SERVING)
clusterv1.RegisterServiceServer(s, &mockServer{
code: code,
+ statusCode: modelv1.Status_STATUS_SUCCEED,
latency: latency,
healthServer: hs,
})
@@ -161,13 +168,6 @@ func setup(address string, code codes.Code, latency time.Duration) func() {
func setupWithStatus(address string, statusCode modelv1.Status) (*health.Server, func()) {
s := grpc.NewServer()
hs := health.NewServer()
- if statusCode == modelv1.Status_STATUS_UNSPECIFIED || statusCode == modelv1.Status_STATUS_SUCCEED {
- hs.SetServingStatus(data.TopicStreamWrite.String(), grpc_health_v1.HealthCheckResponse_SERVING)
- hs.SetServingStatus(data.TopicMeasureWrite.String(), grpc_health_v1.HealthCheckResponse_SERVING)
- } else {
- hs.SetServingStatus(data.TopicStreamWrite.String(), grpc_health_v1.HealthCheckResponse_NOT_SERVING)
- hs.SetServingStatus(data.TopicMeasureWrite.String(), grpc_health_v1.HealthCheckResponse_NOT_SERVING)
- }
clusterv1.RegisterServiceServer(s, &mockServer{
code: codes.OK,
latency: 200 * time.Millisecond,
diff --git a/banyand/queue/pub/pub_test.go b/banyand/queue/pub/pub_test.go
index d7698382..2e90d132 100644
--- a/banyand/queue/pub/pub_test.go
+++ b/banyand/queue/pub/pub_test.go
@@ -29,6 +29,7 @@ import (
"google.golang.org/grpc/codes"
"google.golang.org/grpc/status"
+ "github.com/apache/skywalking-banyandb/api/common"
"github.com/apache/skywalking-banyandb/api/data"
modelv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/model/v1"
streamv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/stream/v1"
@@ -126,7 +127,7 @@ var _ = ginkgo.Describe("Publish and Broadcast", func() {
ginkgo.It("should go to evict queue when node's disk is full", func() {
addr1 := getAddress()
addr2 := getAddress()
- _, closeFn1 := setupWithStatus(addr1, modelv1.Status_STATUS_UNSPECIFIED)
+ _, closeFn1 := setupWithStatus(addr1, modelv1.Status_STATUS_SUCCEED)
_, closeFn2 := setupWithStatus(addr2, modelv1.Status_STATUS_DISK_FULL)
p := newPub()
defer func() {
@@ -146,7 +147,8 @@ var _ = ginkgo.Describe("Publish and Broadcast", func() {
bus.NewBatchMessageWithNode(bus.MessageID(i), "node1", &streamv1.InternalWriteRequest{}),
bus.NewBatchMessageWithNode(bus.MessageID(i), "node2", &streamv1.InternalWriteRequest{}),
)
- gomega.Expect(err).Should(gomega.MatchError("node node2 is not writable"))
+ gomega.Expect(err).Should(gomega.MatchError(common.NewErrorWithStatus(
+ modelv1.Status_STATUS_DISK_FULL, modelv1.Status_name[int32(modelv1.Status_STATUS_DISK_FULL)]).Error()))
}
cee, err := bp.Close()
gomega.Expect(err).ShouldNot(gomega.HaveOccurred())
diff --git a/banyand/queue/queue.go b/banyand/queue/queue.go
index fe59d140..f447d6f3 100644
--- a/banyand/queue/queue.go
+++ b/banyand/queue/queue.go
@@ -56,5 +56,5 @@ type Server interface {
// BatchPublisher is the interface for publishing data in batch.
type BatchPublisher interface {
bus.Publisher
- Close() (map[string]common.Error, error)
+ Close() (map[string]*common.Error, error)
}
diff --git a/banyand/queue/sub/health.go b/banyand/queue/sub/health.go
index 70857204..e2f473d1 100644
--- a/banyand/queue/sub/health.go
+++ b/banyand/queue/sub/health.go
@@ -20,41 +20,29 @@ package sub
import (
"context"
- "google.golang.org/grpc/health"
- healthpb "google.golang.org/grpc/health/grpc_health_v1"
-
- "github.com/apache/skywalking-banyandb/pkg/bus"
+ clusterv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/cluster/v1"
+ modelv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/model/v1"
)
-type healthServer struct {
- *health.Server
- listeners map[string]bus.MessageListener
-}
-
-func newHealthServer(listeners map[bus.Topic]bus.MessageListener) *healthServer {
- s := &healthServer{
- Server: health.NewServer(),
- listeners: make(map[string]bus.MessageListener, len(listeners)),
- }
- for t, l := range listeners {
- topic := t.String()
- s.listeners[topic] = l
- }
- return s
-}
-
-func (s *healthServer) Check(ctx context.Context, in *healthpb.HealthCheckRequest) (*healthpb.HealthCheckResponse, error) {
- if l, ok := s.listeners[in.Service]; ok {
- if l.CheckHealth() != nil {
- s.SetServingStatus(in.Service, healthpb.HealthCheckResponse_NOT_SERVING)
- return &healthpb.HealthCheckResponse{
- Status: healthpb.HealthCheckResponse_NOT_SERVING,
- }, nil
+func (s *server) HealthCheck(_ context.Context, req *clusterv1.HealthCheckRequest) (*clusterv1.HealthCheckResponse, error) {
+ if t, ok := s.topicMap[req.ServiceName]; ok {
+ if l := s.listeners[t]; l != nil {
+ if err := l.CheckHealth(); err != nil {
+ return &clusterv1.HealthCheckResponse{
+ ServiceName: req.ServiceName,
+ Status: err.Status(),
+ Error: err.Error(),
+ }, nil
+ } else {
+ return &clusterv1.HealthCheckResponse{
+ ServiceName: req.ServiceName,
+ Status: modelv1.Status_STATUS_SUCCEED,
+ }, nil
+ }
}
- s.SetServingStatus(in.Service, healthpb.HealthCheckResponse_SERVING)
- return &healthpb.HealthCheckResponse{
- Status: healthpb.HealthCheckResponse_SERVING,
- }, nil
}
- return s.Server.Check(ctx, in)
+ return &clusterv1.HealthCheckResponse{
+ ServiceName: req.ServiceName,
+ Status: modelv1.Status_STATUS_NOT_FOUND,
+ }, nil
}
diff --git a/banyand/queue/sub/server.go b/banyand/queue/sub/server.go
index e4a21cf3..672389a7 100644
--- a/banyand/queue/sub/server.go
+++ b/banyand/queue/sub/server.go
@@ -35,6 +35,7 @@ import (
"google.golang.org/grpc/codes"
"google.golang.org/grpc/credentials"
"google.golang.org/grpc/credentials/insecure"
+ "google.golang.org/grpc/health"
"google.golang.org/grpc/health/grpc_health_v1"
"google.golang.org/grpc/status"
@@ -69,6 +70,7 @@ type server struct {
log *logger.Logger
ser *grpclib.Server
listeners map[bus.Topic]bus.MessageListener
+ topicMap map[string]bus.Topic
clusterv1.UnimplementedServiceServer
metrics *metrics
clientCloser context.CancelFunc
@@ -88,6 +90,7 @@ type server struct {
func NewServer(omr observability.MetricsRegistry) queue.Server {
return &server{
listeners: make(map[bus.Topic]bus.MessageListener),
+ topicMap: make(map[string]bus.Topic),
omr: omr,
}
}
@@ -174,7 +177,7 @@ func (s *server) Serve() run.StopNotify {
)
s.ser = grpclib.NewServer(opts...)
clusterv1.RegisterServiceServer(s.ser, s)
- grpc_health_v1.RegisterHealthServer(s.ser, newHealthServer(s.listeners))
+ grpc_health_v1.RegisterHealthServer(s.ser, health.NewServer())
var ctx context.Context
ctx, s.clientCloser = context.WithCancel(context.Background())
diff --git a/banyand/queue/sub/sub.go b/banyand/queue/sub/sub.go
index 9cf4e787..f61f555f 100644
--- a/banyand/queue/sub/sub.go
+++ b/banyand/queue/sub/sub.go
@@ -72,7 +72,7 @@ func (s *server) Send(stream clusterv1.Service_SendServer) error {
data := message.Data()
if data != nil {
switch d := data.(type) {
- case common.Error:
+ case *common.Error:
resp = &clusterv1.SendResponse{
MessageId: writeEntity.MessageId,
Error: d.Error(),
@@ -158,7 +158,7 @@ func (s *server) Send(stream clusterv1.Service_SendServer) error {
switch d := m.Data().(type) {
case proto.Message:
message = d
- case common.Error:
+ case *common.Error:
select {
case <-ctx.Done():
s.metrics.totalMsgReceivedErr.Inc(1, writeEntity.Topic)
@@ -193,6 +193,7 @@ func (s *server) Subscribe(topic bus.Topic, listener bus.MessageListener) error
defer s.listenersLock.Unlock()
if _, ok := s.listeners[topic]; !ok {
s.listeners[topic] = listener
+ s.topicMap[topic.String()] = topic
return nil
}
return errors.New("topic already exists")
@@ -211,8 +212,8 @@ func (s *server) reply(topic *bus.Topic, stream clusterv1.Service_SendServer, wr
MessageId: writeEntity.MessageId,
}
- // nolint: errorlint
- if ce, ok := err.(common.Error); ok {
+ var ce *common.Error
+ if errors.As(err, &ce) {
resp.Error = ce.Error()
resp.Status = ce.Status()
if topic != nil {
diff --git a/banyand/stream/metadata_test.go b/banyand/stream/metadata_test.go
index 72a78fe4..b77273d5 100644
--- a/banyand/stream/metadata_test.go
+++ b/banyand/stream/metadata_test.go
@@ -323,7 +323,7 @@ func queryAllMeasurements(svcs *services, expectedSize int, newTag []string, new
}
resp = d
return true
- case common.Error:
+ case *common.Error:
Fail(d.Error())
default:
Fail("unexpected data type")
diff --git a/banyand/stream/service.go b/banyand/stream/service.go
index bf477173..18775ea2 100644
--- a/banyand/stream/service.go
+++ b/banyand/stream/service.go
@@ -84,8 +84,8 @@ func (s *service) FlagSet() *run.FlagSet {
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(32 << 20)
- flagS.VarP(&s.option.seriesCacheMaxSize, "measure-series-cache-max-size", "", "the max size of series cache in each group")
- flagS.IntVar(&s.maxDiskUsagePercent, "measure-max-disk-usage-percent", 95, "the maximum disk usage percentage allowed")
+ flagS.VarP(&s.option.seriesCacheMaxSize, "stream-series-cache-max-size", "", "the max size of series cache in each group")
+ flagS.IntVar(&s.maxDiskUsagePercent, "stream-max-disk-usage-percent", 95, "the maximum disk usage percentage allowed")
return flagS
}
@@ -93,6 +93,12 @@ func (s *service) Validate() error {
if s.root == "" {
return errEmptyRootPath
}
+ if s.maxDiskUsagePercent < 0 {
+ return errors.New("stream-max-disk-usage-percent must be greater than or equal to 0")
+ }
+ if s.maxDiskUsagePercent > 100 {
+ return errors.New("stream-max-disk-usage-percent must be less than or equal to 100")
+ }
return nil
}
diff --git a/banyand/stream/write.go b/banyand/stream/write.go
index e1293f0a..db995f6b 100644
--- a/banyand/stream/write.go
+++ b/banyand/stream/write.go
@@ -46,6 +46,9 @@ type writeCallback struct {
}
func setUpWriteCallback(l *logger.Logger, schemaRepo *schemaRepo, maxDiskUsagePercent int) bus.MessageListener {
+ if maxDiskUsagePercent > 100 {
+ maxDiskUsagePercent = 100
+ }
return &writeCallback{
l: l,
schemaRepo: schemaRepo,
@@ -53,9 +56,9 @@ func setUpWriteCallback(l *logger.Logger, schemaRepo *schemaRepo, maxDiskUsagePe
}
}
-func (w *writeCallback) CheckHealth() error {
+func (w *writeCallback) CheckHealth() *common.Error {
if w.maxDiskUsagePercent < 1 {
- return common.NewErrorWithStatus(modelv1.Status_STATUS_DISK_FULL, "measure is readonly because \"measure-max-disk-usage-percent\" is 0")
+ return common.NewErrorWithStatus(modelv1.Status_STATUS_DISK_FULL, "stream is readonly because \"stream-max-disk-usage-percent\" is 0")
}
diskPercent := observability.GetPathUsedPercent(w.schemaRepo.path)
if diskPercent < w.maxDiskUsagePercent {
diff --git a/docs/api-reference.md b/docs/api-reference.md
index 6d5da714..0cda0cd8 100644
--- a/docs/api-reference.md
+++ b/docs/api-reference.md
@@ -7,6 +7,8 @@
- [Status](#banyandb-model-v1-Status)
- [banyandb/cluster/v1/rpc.proto](#banyandb_cluster_v1_rpc-proto)
+ - [HealthCheckRequest](#banyandb-cluster-v1-HealthCheckRequest)
+ - [HealthCheckResponse](#banyandb-cluster-v1-HealthCheckResponse)
- [SendRequest](#banyandb-cluster-v1-SendRequest)
- [SendResponse](#banyandb-cluster-v1-SendResponse)
@@ -264,6 +266,38 @@ Status is the response status for write
+<a name="banyandb-cluster-v1-HealthCheckRequest"></a>
+
+### HealthCheckRequest
+
+
+
+| Field | Type | Label | Description |
+| ----- | ---- | ----- | ----------- |
+| service_name | [string](#string) | | |
+
+
+
+
+
+
+<a name="banyandb-cluster-v1-HealthCheckResponse"></a>
+
+### HealthCheckResponse
+
+
+
+| Field | Type | Label | Description |
+| ----- | ---- | ----- | ----------- |
+| service_name | [string](#string) | | |
+| status | [banyandb.model.v1.Status](#banyandb-model-v1-Status) | | |
+| error | [string](#string) | | |
+
+
+
+
+
+
<a name="banyandb-cluster-v1-SendRequest"></a>
### SendRequest
@@ -315,6 +349,7 @@ Status is the response status for write
| Method Name | Request Type | Response Type | Description |
| ----------- | ------------ | ------------- | ------------|
| Send | [SendRequest](#banyandb-cluster-v1-SendRequest) stream | [SendResponse](#banyandb-cluster-v1-SendResponse) stream | |
+| HealthCheck | [HealthCheckRequest](#banyandb-cluster-v1-HealthCheckRequest) | [HealthCheckResponse](#banyandb-cluster-v1-HealthCheckResponse) | |
@@ -2716,7 +2751,7 @@ WriteResponse is the response contract for write
| Field | Type | Label | Description |
| ----- | ---- | ----- | ----------- |
| message_id | [uint64](#uint64) | | the message_id from request. |
-| status | [banyandb.model.v1.Status](#banyandb-model-v1-Status) | | status indicates the request processing result |
+| status | [string](#string) | | status indicates the request processing result |
| metadata | [banyandb.common.v1.Metadata](#banyandb-common-v1-Metadata) | | the metadata from request when request fails |
@@ -3155,7 +3190,7 @@ QueryResponse is the response for a query to the Query module.
| Field | Type | Label | Description |
| ----- | ---- | ----- | ----------- |
| message_id | [uint64](#uint64) | | the message_id from request. |
-| status | [banyandb.model.v1.Status](#banyandb-model-v1-Status) | | status indicates the request processing result |
+| status | [string](#string) | | status indicates the request processing result |
| metadata | [banyandb.common.v1.Metadata](#banyandb-common-v1-Metadata) | | the metadata from request when request fails |
diff --git a/docs/operation/troubleshooting/overhead.md b/docs/operation/troubleshooting/overhead.md
index 2f99cf1b..a535208b 100644
--- a/docs/operation/troubleshooting/overhead.md
+++ b/docs/operation/troubleshooting/overhead.md
@@ -16,6 +16,10 @@ If you notice high disk usage on the BanyanDB server, follow these steps to trou
1. **Check Group TTL**: Verify that the TTL policy for groups is not causing excessive data storage. If the TTL for a group is set too high, it may result in high disk usage. Use the `bydbctl` command to [update the group schema](../../interacting/bydbctl/schema/group.md#update-operation) and adjust the TTL as needed.
2. **Check Segment Interval**: Check the segment interval for groups to ensure that data is being compacted and stored efficiently. If the TTL is 7 days, the segment interval is set to 3 days. At the 10th morning, the first segment will be deleted. There will be 9 days of data in the database at most, which is more than the TTL.
+## Cannot Write Data
+
+The parameters `measure-max-disk-usage-percent`, `stream-max-disk-usage-percent`, and `property-max-disk-usage-percent` control what percentage of the disk these three different modules can use. If the disk usage exceeds these limits, or if we set these parameters to 0, the module will not accept writing. However, queries can still run. Once the disk usage goes down below the set limits, the module can accept data again.
+
## Too Many Open Files
The BanyanDB uses LSM-tree storage engine, which may open many files. If you encounter issues with too many open files, follow these steps to troubleshoot the issue:
diff --git a/pkg/bus/bus.go b/pkg/bus/bus.go
index 146169d1..7326ca56 100644
--- a/pkg/bus/bus.go
+++ b/pkg/bus/bus.go
@@ -26,6 +26,7 @@ import (
"sync"
"time"
+ "github.com/apache/skywalking-banyandb/api/common"
"go.uber.org/multierr"
)
@@ -88,7 +89,7 @@ func NewMessageWithNode(id MessageID, node string, data interface{}) Message {
// MessageListener is the signature of functions that can handle an EventMessage.
type MessageListener interface {
Rev(ctx context.Context, message Message) Message
- CheckHealth() error
+ CheckHealth() *common.Error
}
// Subscriber allow subscribing a Topic's messages.
@@ -110,7 +111,7 @@ type Broadcaster interface {
type UnImplementedHealthyListener struct{}
// CheckHealth always returns nil.
-func (h *UnImplementedHealthyListener) CheckHealth() error {
+func (h *UnImplementedHealthyListener) CheckHealth() *common.Error {
return nil
}
diff --git a/pkg/bus/bus_test.go b/pkg/bus/bus_test.go
index 67a9444c..f3b7132d 100644
--- a/pkg/bus/bus_test.go
+++ b/pkg/bus/bus_test.go
@@ -25,6 +25,8 @@ import (
"sync"
"testing"
"time"
+
+ "github.com/apache/skywalking-banyandb/api/common"
)
func TestBus_PubAndSub(t *testing.T) {
@@ -263,9 +265,9 @@ type mockListener struct {
unhealthy bool
}
-func (m *mockListener) CheckHealth() error {
+func (m *mockListener) CheckHealth() *common.Error {
if m.unhealthy {
- return errors.New("error")
+ return common.NewError("unhealthy")
}
return nil
}
diff --git a/test/integration/standalone/other/disk_test.go b/test/integration/standalone/other/disk_test.go
index 19fda94e..880ca963 100644
--- a/test/integration/standalone/other/disk_test.go
+++ b/test/integration/standalone/other/disk_test.go
@@ -70,7 +70,7 @@ var _ = g.Describe("Disk", func() {
return
}
gm.Expect(err).ShouldNot(gm.HaveOccurred())
- gm.Expect(resp.GetStatus()).To(gm.Equal(modelv1.Status_STATUS_DISK_FULL))
+ gm.Expect(resp.GetStatus()).To(gm.Equal(modelv1.Status_name[int32(modelv1.Status_STATUS_DISK_FULL)]))
}
})
g.It(" is a cluster, blocking writing on node 0, with disk full", func() {
@@ -113,18 +113,22 @@ var _ = g.Describe("Disk", func() {
wc, deferFn := writeData(liaisonAddr)
defer deferFn()
errNum := 0
+ successNum := 0
for {
resp, err := wc.Recv()
if errors.Is(err, io.EOF) {
break
}
gm.Expect(err).ShouldNot(gm.HaveOccurred())
- if resp.Status != modelv1.Status_STATUS_SUCCEED {
+ if resp.Status == modelv1.Status_name[int32(modelv1.Status_STATUS_SUCCEED)] {
+ successNum++
+ } else {
errNum++
- gm.Expect(resp.GetStatus()).To(gm.Equal(modelv1.Status_STATUS_INTERNAL_ERROR))
+ gm.Expect(resp.GetStatus()).To(gm.Equal(modelv1.Status_name[int32(modelv1.Status_STATUS_DISK_FULL)]))
}
}
gm.Expect(errNum).To(gm.BeNumerically(">", 0))
+ gm.Expect(successNum).To(gm.BeNumerically(">", 0))
})
})