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 2023/08/28 23:40:11 UTC
[skywalking-banyandb] branch main updated: Implement remote queue (#320)
This is an automated email from the ASF dual-hosted git repository.
hanahmily pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/skywalking-banyandb.git
The following commit(s) were added to refs/heads/main by this push:
new 8ee20fda Implement remote queue (#320)
8ee20fda is described below
commit 8ee20fdaaedbc05abbf1d2e225ef1373abcdcfb2
Author: Gao Hongtao <ha...@gmail.com>
AuthorDate: Tue Aug 29 07:32:52 2023 +0800
Implement remote queue (#320)
Signed-off-by: Gao Hongtao <ha...@gmail.com>
---
.golangci.yml | 2 +
CHANGES.md | 1 +
.../proto/banyandb/cluster/v1/rpc.proto | 37 +-
banyand/internal/cmd/liaison.go | 8 +-
banyand/internal/cmd/standalone.go | 5 +-
banyand/internal/cmd/storage.go | 8 +-
banyand/liaison/grpc/discovery.go | 4 +-
banyand/liaison/grpc/measure.go | 7 +-
banyand/liaison/grpc/registry_test.go | 3 +-
banyand/liaison/grpc/server.go | 21 +-
banyand/liaison/grpc/stream.go | 6 +-
banyand/measure/measure_suite_test.go | 3 +-
banyand/measure/service.go | 11 +-
banyand/query/processor.go | 2 +-
banyand/query/query.go | 2 +-
banyand/queue/local.go | 26 +
banyand/queue/pub/client.go | 105 ++++
banyand/queue/pub/pub.go | 186 +++++++
banyand/queue/queue.go | 28 +-
banyand/queue/sub/server.go | 200 ++++++++
banyand/queue/sub/sub.go | 84 +++
banyand/stream/service.go | 4 +-
banyand/stream/stream_suite_test.go | 3 +-
docs/api-reference.md | 564 ++++++++++++---------
pkg/bus/bus.go | 23 +-
pkg/test/setup/setup.go | 3 +-
26 files changed, 1007 insertions(+), 339 deletions(-)
diff --git a/.golangci.yml b/.golangci.yml
index b2ea7995..6c86d559 100644
--- a/.golangci.yml
+++ b/.golangci.yml
@@ -68,6 +68,8 @@ linters-settings:
alias: measurev1
- pkg: github.com/apache/skywalking-banyandb/api/proto/banyandb/stream/v1
alias: streamv1
+ - pkg: github.com/apache/skywalking-banyandb/api/proto/banyandb/cluster/v1
+ alias: clusterv1
- pkg: github.com/apache/skywalking-banyandb/pkg/pb/v1
alias: pbv1
lll:
diff --git a/CHANGES.md b/CHANGES.md
index 92c6e2fd..667cdb6a 100644
--- a/CHANGES.md
+++ b/CHANGES.md
@@ -12,6 +12,7 @@ Release Notes.
- Support multiple roles for banyand server.
- Support for recovery buffer using wal.
- Register the node role to the metadata registry.
+- Implement the remote queue to spreading data to data nodes.
### Bugs
diff --git a/banyand/queue/queue.go b/api/proto/banyandb/cluster/v1/rpc.proto
similarity index 58%
copy from banyand/queue/queue.go
copy to api/proto/banyandb/cluster/v1/rpc.proto
index e4bc0b33..076c21e6 100644
--- a/banyand/queue/queue.go
+++ b/api/proto/banyandb/cluster/v1/rpc.proto
@@ -15,29 +15,26 @@
// specific language governing permissions and limitations
// under the License.
-package queue
+syntax = "proto3";
-import (
- "context"
+package banyandb.cluster.v1;
- "github.com/apache/skywalking-banyandb/pkg/bus"
- "github.com/apache/skywalking-banyandb/pkg/run"
-)
+import "banyandb/measure/v1/write.proto";
+import "banyandb/stream/v1/write.proto";
-// Queue builds a data transmission tunnel between subscribers and publishers.
-//
-//go:generate mockgen -destination=./queue_mock.go -package=queue github.com/apache/skywalking-banyandb/pkg/bus MessageListener
-type Queue interface {
- run.Unit
- bus.Subscriber
- bus.Publisher
- run.Service
+option go_package = "github.com/apache/skywalking-banyandb/api/proto/banyandb/cluster/v1";
+
+message WriteRequest {
+ string topic = 1;
+ uint64 message_id = 2;
+ oneof request {
+ stream.v1.InternalWriteRequest stream = 3;
+ measure.v1.InternalWriteRequest measure = 4;
+ }
}
-// NewQueue return a new Queue.
-func NewQueue(_ context.Context) (Queue, error) {
- return &local{
- local: bus.NewBus(),
- stopCh: make(chan struct{}),
- }, nil
+message WriteResponse {}
+
+service Service {
+ rpc Write(stream WriteRequest) returns (stream WriteResponse);
}
diff --git a/banyand/internal/cmd/liaison.go b/banyand/internal/cmd/liaison.go
index d8847c7c..0b842201 100644
--- a/banyand/internal/cmd/liaison.go
+++ b/banyand/internal/cmd/liaison.go
@@ -28,7 +28,7 @@ import (
"github.com/apache/skywalking-banyandb/banyand/liaison/http"
"github.com/apache/skywalking-banyandb/banyand/metadata"
"github.com/apache/skywalking-banyandb/banyand/observability"
- "github.com/apache/skywalking-banyandb/banyand/queue"
+ "github.com/apache/skywalking-banyandb/banyand/queue/pub"
"github.com/apache/skywalking-banyandb/pkg/logger"
"github.com/apache/skywalking-banyandb/pkg/run"
"github.com/apache/skywalking-banyandb/pkg/signal"
@@ -40,15 +40,11 @@ var liaisonGroup = run.NewGroup("liaison")
func newLiaisonCmd() *cobra.Command {
l := logger.GetLogger("bootstrap")
ctx := context.Background()
- // nolint: staticcheck
- pipeline, err := queue.NewQueue(ctx)
- if err != nil {
- l.Fatal().Err(err).Msg("failed to initiate data pipeline")
- }
metaSvc, err := metadata.NewClient(ctx)
if err != nil {
l.Fatal().Err(err).Msg("failed to initiate metadata service")
}
+ pipeline := pub.New(metaSvc)
grpcServer := grpc.NewServer(ctx, pipeline, metaSvc)
profSvc := observability.NewProfService()
metricSvc := observability.NewMetricService()
diff --git a/banyand/internal/cmd/standalone.go b/banyand/internal/cmd/standalone.go
index f45a68d3..8686db07 100644
--- a/banyand/internal/cmd/standalone.go
+++ b/banyand/internal/cmd/standalone.go
@@ -43,10 +43,7 @@ var standaloneGroup = run.NewGroup("standalone")
func newStandaloneCmd() *cobra.Command {
l := logger.GetLogger("bootstrap")
ctx := context.Background()
- pipeline, err := queue.NewQueue(ctx)
- if err != nil {
- l.Fatal().Err(err).Msg("failed to initiate data pipeline")
- }
+ pipeline := queue.Local()
metaSvc, err := metadata.NewService(ctx)
if err != nil {
l.Fatal().Err(err).Msg("failed to initiate metadata service")
diff --git a/banyand/internal/cmd/storage.go b/banyand/internal/cmd/storage.go
index a8f4e159..ceadaa17 100644
--- a/banyand/internal/cmd/storage.go
+++ b/banyand/internal/cmd/storage.go
@@ -28,7 +28,7 @@ import (
"github.com/apache/skywalking-banyandb/banyand/metadata"
"github.com/apache/skywalking-banyandb/banyand/observability"
"github.com/apache/skywalking-banyandb/banyand/query"
- "github.com/apache/skywalking-banyandb/banyand/queue"
+ "github.com/apache/skywalking-banyandb/banyand/queue/sub"
"github.com/apache/skywalking-banyandb/banyand/stream"
"github.com/apache/skywalking-banyandb/pkg/config"
"github.com/apache/skywalking-banyandb/pkg/logger"
@@ -50,15 +50,11 @@ var flagStorageMode string
func newStorageCmd() *cobra.Command {
l := logger.GetLogger("bootstrap")
ctx := context.Background()
- // nolint: staticcheck
- pipeline, err := queue.NewQueue(ctx)
- if err != nil {
- l.Fatal().Err(err).Msg("failed to initiate data pipeline")
- }
metaSvc, err := metadata.NewClient(ctx)
if err != nil {
l.Fatal().Err(err).Msg("failed to initiate metadata service")
}
+ pipeline := sub.NewServer()
streamSvc, err := stream.NewService(ctx, metaSvc, pipeline)
if err != nil {
l.Fatal().Err(err).Msg("failed to initiate stream service")
diff --git a/banyand/liaison/grpc/discovery.go b/banyand/liaison/grpc/discovery.go
index ad0b3f58..e2fb24c9 100644
--- a/banyand/liaison/grpc/discovery.go
+++ b/banyand/liaison/grpc/discovery.go
@@ -40,7 +40,7 @@ import (
var errNotExist = errors.New("the object doesn't exist")
type discoveryService struct {
- pipeline queue.Queue
+ pipeline queue.Client
metadataRepo metadata.Repo
shardRepo *shardRepo
entityRepo *entityRepo
@@ -48,7 +48,7 @@ type discoveryService struct {
kind schema.Kind
}
-func newDiscoveryService(pipeline queue.Queue, kind schema.Kind, metadataRepo metadata.Repo) *discoveryService {
+func newDiscoveryService(pipeline queue.Client, kind schema.Kind, metadataRepo metadata.Repo) *discoveryService {
sr := &shardRepo{shardEventsMap: make(map[identity]uint32)}
er := &entityRepo{entitiesMap: make(map[identity]partition.EntityLocator)}
return &discoveryService{
diff --git a/banyand/liaison/grpc/measure.go b/banyand/liaison/grpc/measure.go
index f53009a6..8a541147 100644
--- a/banyand/liaison/grpc/measure.go
+++ b/banyand/liaison/grpc/measure.go
@@ -62,6 +62,8 @@ func (ms *measureService) Write(measure measurev1.MeasureService_WriteServer) er
}
}
ctx := measure.Context()
+ publisher := ms.pipeline.NewBatchPublisher()
+ defer publisher.Close()
for {
select {
case <-ctx.Done():
@@ -99,8 +101,9 @@ func (ms *measureService) Write(measure measurev1.MeasureService_WriteServer) er
SeriesHash: tsdb.HashEntity(entity),
EntityValues: tagValues.Encode(),
}
- message := bus.NewMessage(bus.MessageID(time.Now().UnixNano()), iwr)
- _, errWritePub := ms.pipeline.Publish(data.TopicMeasureWrite, message)
+ // TODO: set node id
+ message := bus.NewMessageWithNode(bus.MessageID(time.Now().UnixNano()), "todo", iwr)
+ _, errWritePub := publisher.Publish(data.TopicMeasureWrite, message)
if errWritePub != nil {
ms.sampled.Error().Err(errWritePub).RawJSON("written", logger.Proto(writeRequest)).Msg("failed to send a message")
}
diff --git a/banyand/liaison/grpc/registry_test.go b/banyand/liaison/grpc/registry_test.go
index 213f5242..6c86d780 100644
--- a/banyand/liaison/grpc/registry_test.go
+++ b/banyand/liaison/grpc/registry_test.go
@@ -173,8 +173,7 @@ var _ = Describe("Registry", func() {
func setupForRegistry() func() {
// Init `Queue` module
- pipeline, err := queue.NewQueue(context.TODO())
- Expect(err).NotTo(HaveOccurred())
+ pipeline := queue.Local()
// Init `Metadata` module
metaSvc, err := metadata.NewService(context.TODO())
Expect(err).NotTo(HaveOccurred())
diff --git a/banyand/liaison/grpc/server.go b/banyand/liaison/grpc/server.go
index eae908f9..cc7e0817 100644
--- a/banyand/liaison/grpc/server.go
+++ b/banyand/liaison/grpc/server.go
@@ -57,23 +57,6 @@ var (
errAccessLogRootPath = errors.New("access log root path is required")
)
-// // 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.
-
// Server defines the gRPC server.
type Server interface {
run.Unit
@@ -81,7 +64,7 @@ type Server interface {
}
type server struct {
- pipeline queue.Queue
+ pipeline queue.Client
creds credentials.TransportCredentials
*streamRegistryServer
log *logger.Logger
@@ -108,7 +91,7 @@ type server struct {
}
// NewServer returns a new gRPC server.
-func NewServer(_ context.Context, pipeline queue.Queue, schemaRegistry metadata.Repo) Server {
+func NewServer(_ context.Context, pipeline queue.Client, schemaRegistry metadata.Repo) Server {
streamSVC := &streamService{
discoveryService: newDiscoveryService(pipeline, schema.KindStream, schemaRegistry),
}
diff --git a/banyand/liaison/grpc/stream.go b/banyand/liaison/grpc/stream.go
index 0ff834ec..e4cc1ed1 100644
--- a/banyand/liaison/grpc/stream.go
+++ b/banyand/liaison/grpc/stream.go
@@ -61,6 +61,8 @@ func (s *streamService) Write(stream streamv1.StreamService_WriteServer) error {
logger.Err(errResp).Msg("failed to send response")
}
}
+ publisher := s.pipeline.NewBatchPublisher()
+ defer publisher.Close()
ctx := stream.Context()
for {
select {
@@ -101,8 +103,8 @@ func (s *streamService) Write(stream streamv1.StreamService_WriteServer) error {
if s.log.Debug().Enabled() {
iwr.EntityValues = tagValues.Encode()
}
- message := bus.NewMessage(bus.MessageID(time.Now().UnixNano()), iwr)
- _, errWritePub := s.pipeline.Publish(data.TopicStreamWrite, message)
+ message := bus.NewMessageWithNode(bus.MessageID(time.Now().UnixNano()), "TODO", iwr)
+ _, errWritePub := publisher.Publish(data.TopicStreamWrite, message)
if errWritePub != nil {
s.sampled.Error().Err(errWritePub).RawJSON("written", logger.Proto(writeEntity)).Msg("failed to send a message")
}
diff --git a/banyand/measure/measure_suite_test.go b/banyand/measure/measure_suite_test.go
index 25e31fd9..10c3c6fc 100644
--- a/banyand/measure/measure_suite_test.go
+++ b/banyand/measure/measure_suite_test.go
@@ -69,8 +69,7 @@ func setUp() (*services, func()) {
gomega.Expect(ctrl).ShouldNot(gomega.BeNil())
// Init Pipeline
- pipeline, err := queue.NewQueue(context.TODO())
- gomega.Expect(err).NotTo(gomega.HaveOccurred())
+ pipeline := queue.Local()
// Init Metadata Service
metadataService, err := metadata.NewService(context.TODO())
diff --git a/banyand/measure/service.go b/banyand/measure/service.go
index 2ff76731..47c93684 100644
--- a/banyand/measure/service.go
+++ b/banyand/measure/service.go
@@ -57,7 +57,8 @@ type service struct {
schemaRepo schemaRepo
writeListener bus.MessageListener
metadata metadata.Repo
- pipeline queue.Queue
+ pipeline queue.Server
+ localPipeline queue.Queue
l *logger.Logger
root string
dbOpts tsdb.DatabaseOpts
@@ -109,8 +110,9 @@ func (s *service) PreRun(_ context.Context) error {
s.l = logger.GetLogger(s.Name())
path := path.Join(s.root, s.Name())
observability.UpdatePath(path)
+ s.localPipeline = queue.Local()
s.schemaRepo = newSchemaRepo(path, s.metadata, s.dbOpts,
- s.l, s.pipeline, int64(s.BlockEncoderBufferSize), int64(s.BlockBufferSize))
+ s.l, s.localPipeline, int64(s.BlockEncoderBufferSize), int64(s.BlockBufferSize))
// run a serial watcher
go s.schemaRepo.Watcher()
s.metadata.
@@ -121,7 +123,7 @@ func (s *service) PreRun(_ context.Context) error {
if err != nil {
return err
}
- return nil
+ return s.localPipeline.Subscribe(data.TopicMeasureWrite, s.writeListener)
}
func (s *service) Serve() run.StopNotify {
@@ -129,11 +131,12 @@ func (s *service) Serve() run.StopNotify {
}
func (s *service) GracefulStop() {
+ s.localPipeline.GracefulStop()
s.schemaRepo.Close()
}
// NewService returns a new service.
-func NewService(_ context.Context, metadata metadata.Repo, pipeline queue.Queue) (Service, error) {
+func NewService(_ context.Context, metadata metadata.Repo, pipeline queue.Server) (Service, error) {
return &service{
metadata: metadata,
pipeline: pipeline,
diff --git a/banyand/query/processor.go b/banyand/query/processor.go
index cf36474a..67df4b56 100644
--- a/banyand/query/processor.go
+++ b/banyand/query/processor.go
@@ -55,7 +55,7 @@ type queryService struct {
log *logger.Logger
// TODO: remove the metaService once https://github.com/apache/skywalking/issues/10121 is fixed.
metaService metadata.Repo
- pipeline queue.Queue
+ pipeline queue.Server
sqp *streamQueryProcessor
mqp *measureQueryProcessor
tqp *topNQueryProcessor
diff --git a/banyand/query/query.go b/banyand/query/query.go
index fbff5030..396fe2a0 100644
--- a/banyand/query/query.go
+++ b/banyand/query/query.go
@@ -30,7 +30,7 @@ import (
// NewService return a new query service.
func NewService(_ context.Context, streamService stream.Service, measureService measure.Service,
- metaService metadata.Repo, pipeline queue.Queue,
+ metaService metadata.Repo, pipeline queue.Server,
) (run.Unit, error) {
svc := &queryService{
metaService: metaService,
diff --git a/banyand/queue/local.go b/banyand/queue/local.go
index 14a8632b..659e908e 100644
--- a/banyand/queue/local.go
+++ b/banyand/queue/local.go
@@ -33,6 +33,14 @@ type local struct {
stopCh chan struct{}
}
+// Local return a new local Queue.
+func Local() Queue {
+ return &local{
+ local: bus.NewBus(),
+ stopCh: make(chan struct{}),
+ }
+}
+
// GracefulStop implements Queue.
func (l *local) GracefulStop() {
l.local.Close()
@@ -57,3 +65,21 @@ func (l *local) Publish(topic bus.Topic, message ...bus.Message) (bus.Future, er
func (l local) Name() string {
return "local-pipeline"
}
+
+func (l local) NewBatchPublisher() BatchPublisher {
+ return &localBatchPublisher{
+ local: l.local,
+ }
+}
+
+type localBatchPublisher struct {
+ local *bus.Bus
+}
+
+func (l *localBatchPublisher) Publish(topic bus.Topic, message ...bus.Message) (bus.Future, error) {
+ return l.local.Publish(topic, message...)
+}
+
+func (l *localBatchPublisher) Close() error {
+ return nil
+}
diff --git a/banyand/queue/pub/client.go b/banyand/queue/pub/client.go
new file mode 100644
index 00000000..8fb06418
--- /dev/null
+++ b/banyand/queue/pub/client.go
@@ -0,0 +1,105 @@
+// 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 pub
+
+import (
+ "google.golang.org/grpc"
+ "google.golang.org/grpc/credentials/insecure"
+
+ clusterv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/cluster/v1"
+ databasev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/database/v1"
+ "github.com/apache/skywalking-banyandb/banyand/metadata/schema"
+)
+
+var retryPolicy = `{
+ "methodConfig": [{
+ "name": [{"service": "grpc.examples.echo.Echo"}],
+ "waitForReady": true,
+ "retryPolicy": {
+ "MaxAttempts": 4,
+ "InitialBackoff": ".5s",
+ "MaxBackoff": "10s",
+ "BackoffMultiplier": 1.0,
+ "RetryableStatusCodes": [ "UNAVAILABLE" ]
+ }
+ }]}`
+
+type client struct {
+ client clusterv1.ServiceClient
+ conn *grpc.ClientConn
+}
+
+func (p *pub) OnAddOrUpdate(md schema.Metadata) {
+ if md.Kind != schema.KindNode {
+ return
+ }
+ node, ok := md.Spec.(*databasev1.Node)
+ if !ok {
+ p.log.Warn().Msg("failed to cast node spec")
+ return
+ }
+ address := node.GrpcAddress
+ if address == "" {
+ p.log.Warn().Stringer("node", node).Msg("grpc address is empty")
+ return
+ }
+ name := node.Metadata.GetName()
+ if name == "" {
+ p.log.Warn().Stringer("node", node).Msg("node name is empty")
+ return
+ }
+ p.mu.Lock()
+ defer p.mu.Unlock()
+
+ // If the client already exists, just return
+ if _, ok := p.clients[name]; ok {
+ return
+ }
+ conn, err := grpc.Dial(address, grpc.WithTransportCredentials(insecure.NewCredentials()), grpc.WithDefaultServiceConfig(retryPolicy), grpc.WithBlock())
+ if err != nil {
+ p.log.Error().Err(err).Msg("failed to connect to grpc server")
+ return
+ }
+ c := clusterv1.NewServiceClient(conn)
+ p.clients[name] = &client{conn: conn, client: c}
+}
+
+func (p *pub) OnDelete(md schema.Metadata) {
+ if md.Kind != schema.KindNode {
+ return
+ }
+ node, ok := md.Spec.(*databasev1.Node)
+ if !ok {
+ p.log.Warn().Msg("failed to cast node spec")
+ return
+ }
+ name := node.Metadata.GetName()
+ if name == "" {
+ p.log.Warn().Stringer("node", node).Msg("node name is empty")
+ return
+ }
+ p.mu.Lock()
+ defer p.mu.Unlock()
+
+ if client, ok := p.clients[name]; ok {
+ if client.conn != nil {
+ client.conn.Close() // Close the client connection
+ }
+ delete(p.clients, name)
+ }
+}
diff --git a/banyand/queue/pub/pub.go b/banyand/queue/pub/pub.go
new file mode 100644
index 00000000..1f3e3cf4
--- /dev/null
+++ b/banyand/queue/pub/pub.go
@@ -0,0 +1,186 @@
+// 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 pub implements the queue client.
+package pub
+
+import (
+ "context"
+ "fmt"
+ "sync"
+ "time"
+
+ "go.uber.org/multierr"
+
+ clusterv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/cluster/v1"
+ measurev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/measure/v1"
+ streamv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/stream/v1"
+ "github.com/apache/skywalking-banyandb/banyand/metadata"
+ "github.com/apache/skywalking-banyandb/banyand/metadata/schema"
+ "github.com/apache/skywalking-banyandb/banyand/queue"
+ "github.com/apache/skywalking-banyandb/pkg/bus"
+ "github.com/apache/skywalking-banyandb/pkg/logger"
+ "github.com/apache/skywalking-banyandb/pkg/run"
+)
+
+var (
+ _ run.PreRunner = (*pub)(nil)
+ _ run.Service = (*pub)(nil)
+)
+
+type pub struct {
+ metadata metadata.Repo
+ log *logger.Logger
+ clients map[string]*client
+ closer *run.Closer
+ mu sync.RWMutex
+}
+
+// GracefulStop implements run.Service.
+func (p *pub) GracefulStop() {
+ p.closer.Done()
+ p.closer.CloseThenWait()
+ p.mu.Lock()
+ defer p.mu.Unlock()
+ for _, c := range p.clients {
+ _ = c.conn.Close()
+ }
+}
+
+// Serve implements run.Service.
+func (p *pub) Serve() run.StopNotify {
+ return p.closer.CloseNotify()
+}
+
+func (*pub) Publish(_ bus.Topic, _ ...bus.Message) (bus.Future, error) {
+ panic("unimplemented")
+}
+
+// NewBatchPublisher returns a new batch publisher.
+func (p *pub) NewBatchPublisher() queue.BatchPublisher {
+ return &batchPublisher{pub: p, streams: make(map[string]writeStream)}
+}
+
+// New returns a new queue client.
+func New(metadata metadata.Repo) queue.Client {
+ return &pub{
+ metadata: metadata,
+ closer: run.NewCloser(1),
+ }
+}
+
+// Name implements run.PreRunner.
+func (*pub) Name() string {
+ return "queue-client"
+}
+
+// PreRun implements run.PreRunner.
+func (p *pub) PreRun(context.Context) error {
+ p.log = logger.GetLogger("server-queue")
+ p.metadata.RegisterHandler("queue-client", schema.KindNode, p)
+ return nil
+}
+
+type writeStream struct {
+ client clusterv1.Service_WriteClient
+ cancel func()
+}
+
+type batchPublisher struct {
+ pub *pub
+ streams map[string]writeStream
+}
+
+func (bp *batchPublisher) Close() (err error) {
+ for _, ws := range bp.streams {
+ err = multierr.Append(err, ws.client.CloseSend())
+ ws.cancel()
+ }
+ return err
+}
+
+func (bp *batchPublisher) Publish(topic bus.Topic, message ...bus.Message) (bus.Future, error) {
+ var err error
+ for _, m := range message {
+ if !m.IsRemote() {
+ continue
+ }
+ r := &clusterv1.WriteRequest{
+ Topic: topic.String(),
+ MessageId: uint64(m.ID()),
+ }
+ switch d := m.Data().(type) {
+ case *streamv1.InternalWriteRequest:
+ r.Request = &clusterv1.WriteRequest_Stream{Stream: d}
+ case *measurev1.InternalWriteRequest:
+ r.Request = &clusterv1.WriteRequest_Measure{Measure: d}
+ default:
+ err = multierr.Append(err, fmt.Errorf("invalid message type %T", d))
+ continue
+ }
+ node := m.Node()
+
+ sendData := func() (success bool) {
+ if stream, ok := bp.streams[node]; !ok {
+ defer func() {
+ if !success {
+ delete(bp.streams, node)
+ stream.cancel()
+ }
+ }()
+ select {
+ case <-stream.client.Context().Done():
+ return false
+ default:
+ }
+ errSend := stream.client.Send(r)
+ if errSend != nil {
+ err = multierr.Append(errSend, fmt.Errorf("failed to send message to node %s: %w", node, err))
+ return false
+ }
+ _, errRecv := stream.client.Recv()
+ return errRecv == nil
+ }
+ return false
+ }
+ if sendData() {
+ continue
+ }
+
+ bp.pub.mu.Lock()
+ client, ok := bp.pub.clients[node]
+ bp.pub.mu.Unlock()
+ if !ok {
+ err = multierr.Append(err, fmt.Errorf("failed to get client for node %s", node))
+ continue
+ }
+ //nolint: govet
+ ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+ stream, errCreateStream := client.client.Write(ctx)
+ if err != nil {
+ err = multierr.Append(err, fmt.Errorf("failed to get stream for node %s: %w", node, errCreateStream))
+ continue
+ }
+ bp.streams[node] = writeStream{
+ client: stream,
+ cancel: cancel,
+ }
+ _ = sendData()
+ }
+ //nolint: govet
+ return nil, err
+}
diff --git a/banyand/queue/queue.go b/banyand/queue/queue.go
index e4bc0b33..d899162e 100644
--- a/banyand/queue/queue.go
+++ b/banyand/queue/queue.go
@@ -18,7 +18,7 @@
package queue
import (
- "context"
+ "io"
"github.com/apache/skywalking-banyandb/pkg/bus"
"github.com/apache/skywalking-banyandb/pkg/run"
@@ -28,16 +28,26 @@ import (
//
//go:generate mockgen -destination=./queue_mock.go -package=queue github.com/apache/skywalking-banyandb/pkg/bus MessageListener
type Queue interface {
+ Client
+ Server
+ run.Service
+}
+
+// Client is the interface for publishing data to the queue.
+type Client interface {
run.Unit
- bus.Subscriber
bus.Publisher
- run.Service
+ NewBatchPublisher() BatchPublisher
}
-// NewQueue return a new Queue.
-func NewQueue(_ context.Context) (Queue, error) {
- return &local{
- local: bus.NewBus(),
- stopCh: make(chan struct{}),
- }, nil
+// Server is the interface for receiving data from the queue.
+type Server interface {
+ run.Unit
+ bus.Subscriber
+}
+
+// BatchPublisher is the interface for publishing data in batch.
+type BatchPublisher interface {
+ bus.Publisher
+ io.Closer
}
diff --git a/banyand/queue/sub/server.go b/banyand/queue/sub/server.go
new file mode 100644
index 00000000..d7e415da
--- /dev/null
+++ b/banyand/queue/sub/server.go
@@ -0,0 +1,200 @@
+// 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 sub
+
+import (
+ "context"
+ "net"
+ "runtime/debug"
+ "strconv"
+ "sync"
+ "time"
+
+ "github.com/grpc-ecosystem/go-grpc-middleware/v2/interceptors/recovery"
+ grpc_validator "github.com/grpc-ecosystem/go-grpc-middleware/v2/interceptors/validator"
+ "github.com/pkg/errors"
+ grpclib "google.golang.org/grpc"
+ "google.golang.org/grpc/codes"
+ "google.golang.org/grpc/credentials"
+ "google.golang.org/grpc/health"
+ "google.golang.org/grpc/health/grpc_health_v1"
+ "google.golang.org/grpc/status"
+
+ clusterv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/cluster/v1"
+ databasev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/database/v1"
+ "github.com/apache/skywalking-banyandb/banyand/observability"
+ "github.com/apache/skywalking-banyandb/banyand/queue"
+ "github.com/apache/skywalking-banyandb/pkg/bus"
+ "github.com/apache/skywalking-banyandb/pkg/logger"
+ "github.com/apache/skywalking-banyandb/pkg/run"
+)
+
+const defaultRecvSize = 10 << 20
+
+var (
+ errServerCert = errors.New("invalid server cert file")
+ errServerKey = errors.New("invalid server key file")
+ errNoAddr = errors.New("no address")
+
+ _ run.PreRunner = (*server)(nil)
+ _ run.Service = (*server)(nil)
+)
+
+type server struct {
+ creds credentials.TransportCredentials
+ log *logger.Logger
+ ser *grpclib.Server
+ listeners map[bus.Topic][]bus.MessageListener
+ *clusterv1.UnimplementedServiceServer
+ addr string
+ certFile string
+ keyFile string
+ host string
+ maxRecvMsgSize run.Bytes
+ listenersLock sync.RWMutex
+ port uint32
+ tls bool
+}
+
+// NewServer returns a new gRPC server.
+func NewServer() queue.Server {
+ return &server{}
+}
+
+func (s *server) PreRun(_ context.Context) error {
+ s.log = logger.GetLogger("server-queue")
+ return nil
+}
+
+func (s *server) Name() string {
+ return "server-queue"
+}
+
+func (s *server) Role() databasev1.Role {
+ return databasev1.Role_ROLE_DATA
+}
+
+func (s *server) GetPort() *uint32 {
+ return &s.port
+}
+
+func (s *server) FlagSet() *run.FlagSet {
+ fs := run.NewFlagSet("grpc")
+ s.maxRecvMsgSize = defaultRecvSize
+ fs.VarP(&s.maxRecvMsgSize, "max-recv-msg-size", "", "the size of max receiving message")
+ fs.BoolVar(&s.tls, "tls", false, "connection uses TLS if true, else plain TCP")
+ fs.StringVar(&s.certFile, "cert-file", "", "the TLS cert file")
+ fs.StringVar(&s.keyFile, "key-file", "", "the TLS key file")
+ fs.StringVar(&s.host, "grpc-host", "", "the host of banyand listens")
+ fs.Uint32Var(&s.port, "grpc-port", 17912, "the port of banyand listens")
+ return fs
+}
+
+func (s *server) Validate() error {
+ s.addr = net.JoinHostPort(s.host, strconv.FormatUint(uint64(s.port), 10))
+ if s.addr == ":" {
+ return errNoAddr
+ }
+ observability.UpdateAddress("grpc", s.addr)
+ if !s.tls {
+ return nil
+ }
+ if s.certFile == "" {
+ return errServerCert
+ }
+ if s.keyFile == "" {
+ return errServerKey
+ }
+ creds, errTLS := credentials.NewServerTLSFromFile(s.certFile, s.keyFile)
+ if errTLS != nil {
+ return errors.Wrap(errTLS, "failed to load cert and key")
+ }
+ s.creds = creds
+ return nil
+}
+
+func (s *server) Serve() run.StopNotify {
+ var opts []grpclib.ServerOption
+ if s.tls {
+ opts = []grpclib.ServerOption{grpclib.Creds(s.creds)}
+ }
+ grpcPanicRecoveryHandler := func(p any) (err error) {
+ s.log.Error().Interface("panic", p).Str("stack", string(debug.Stack())).Msg("recovered from panic")
+
+ return status.Errorf(codes.Internal, "%s", p)
+ }
+
+ unaryMetrics, streamMetrics := observability.MetricsServerInterceptor()
+ streamChain := []grpclib.StreamServerInterceptor{
+ recovery.StreamServerInterceptor(recovery.WithRecoveryHandler(grpcPanicRecoveryHandler)),
+ }
+ if streamMetrics != nil {
+ streamChain = append(streamChain, streamMetrics)
+ }
+ unaryChain := []grpclib.UnaryServerInterceptor{
+ grpc_validator.UnaryServerInterceptor(),
+ recovery.UnaryServerInterceptor(recovery.WithRecoveryHandler(grpcPanicRecoveryHandler)),
+ }
+ if unaryMetrics != nil {
+ unaryChain = append(unaryChain, unaryMetrics)
+ }
+
+ opts = append(opts, grpclib.MaxRecvMsgSize(int(s.maxRecvMsgSize)),
+ grpclib.ChainUnaryInterceptor(unaryChain...),
+ grpclib.ChainStreamInterceptor(streamChain...),
+ )
+ s.ser = grpclib.NewServer(opts...)
+ clusterv1.RegisterServiceServer(s.ser, s)
+ grpc_health_v1.RegisterHealthServer(s.ser, health.NewServer())
+
+ stopCh := make(chan struct{})
+ go func(stopCh chan struct{}) {
+ lis, err := net.Listen("tcp", s.addr)
+ if err != nil {
+ s.log.Error().Err(err).Msg("Failed to listen")
+ close(stopCh)
+ return
+ }
+ s.log.Info().Str("addr", s.addr).Msg("Listening to")
+ err = s.ser.Serve(lis)
+ if err != nil {
+ s.log.Error().Err(err).Msg("server is interrupted")
+ }
+ close(stopCh)
+ }(stopCh)
+ return stopCh
+}
+
+func (s *server) GracefulStop() {
+ s.log.Info().Msg("stopping")
+ stopped := make(chan struct{})
+ go func() {
+ s.ser.GracefulStop()
+ close(stopped)
+ }()
+
+ t := time.NewTimer(10 * time.Second)
+ select {
+ case <-t.C:
+ s.ser.Stop()
+ s.log.Info().Msg("force stopped")
+ case <-stopped:
+ t.Stop()
+ s.log.Info().Msg("stopped gracefully")
+ }
+}
diff --git a/banyand/queue/sub/sub.go b/banyand/queue/sub/sub.go
new file mode 100644
index 00000000..d53d4fe1
--- /dev/null
+++ b/banyand/queue/sub/sub.go
@@ -0,0 +1,84 @@
+// 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 sub implements the queue server.
+package sub
+
+import (
+ "io"
+
+ "github.com/pkg/errors"
+
+ clusterv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/cluster/v1"
+ "github.com/apache/skywalking-banyandb/pkg/bus"
+)
+
+// Write implements v1.ServiceServer.
+func (s *server) Write(stream clusterv1.Service_WriteServer) error {
+ reply := func(stream clusterv1.Service_WriteServer) {
+ if errResp := stream.Send(&clusterv1.WriteResponse{}); errResp != nil {
+ s.log.Err(errResp).Msg("failed to send response")
+ }
+ }
+ ctx := stream.Context()
+ var topic *bus.Topic
+ for {
+ select {
+ case <-ctx.Done():
+ return ctx.Err()
+ default:
+ }
+ writeEntity, err := stream.Recv()
+ if errors.Is(err, io.EOF) {
+ return nil
+ }
+ if err != nil {
+ s.log.Error().Stringer("written", writeEntity).Err(err).Msg("failed to receive message")
+ reply(stream)
+ continue
+ }
+ if writeEntity.Topic != "" && topic == nil {
+ t := bus.UniTopic(writeEntity.Topic)
+ topic = &t
+ }
+ if topic == nil {
+ s.log.Error().Stringer("written", writeEntity).Msg("topic is empty")
+ reply(stream)
+ continue
+ }
+ for _, listener := range s.getListeners(*topic) {
+ _ = listener.Rev(bus.NewMessage(bus.MessageID(writeEntity.MessageId), writeEntity.Request))
+ }
+ reply(stream)
+ }
+}
+
+func (s *server) Subscribe(topic bus.Topic, listener bus.MessageListener) error {
+ s.listenersLock.Lock()
+ defer s.listenersLock.Unlock()
+ if _, ok := s.listeners[topic]; !ok {
+ s.listeners[topic] = make([]bus.MessageListener, 0)
+ }
+ s.listeners[topic] = append(s.listeners[topic], listener)
+ return nil
+}
+
+func (s *server) getListeners(topic bus.Topic) []bus.MessageListener {
+ s.listenersLock.RLock()
+ defer s.listenersLock.RUnlock()
+ return s.listeners[topic]
+}
diff --git a/banyand/stream/service.go b/banyand/stream/service.go
index b9fd39b3..3af678d0 100644
--- a/banyand/stream/service.go
+++ b/banyand/stream/service.go
@@ -53,7 +53,7 @@ var _ Service = (*service)(nil)
type service struct {
schemaRepo schemaRepo
metadata metadata.Repo
- pipeline queue.Queue
+ pipeline queue.Server
writeListener *writeCallback
l *logger.Logger
root string
@@ -125,7 +125,7 @@ func (s *service) GracefulStop() {
}
// NewService returns a new service.
-func NewService(_ context.Context, metadata metadata.Repo, pipeline queue.Queue) (Service, error) {
+func NewService(_ context.Context, metadata metadata.Repo, pipeline queue.Server) (Service, error) {
return &service{
metadata: metadata,
pipeline: pipeline,
diff --git a/banyand/stream/stream_suite_test.go b/banyand/stream/stream_suite_test.go
index 898da965..82cb06b0 100644
--- a/banyand/stream/stream_suite_test.go
+++ b/banyand/stream/stream_suite_test.go
@@ -66,8 +66,7 @@ func setUp() (*services, func()) {
ctrl := gomock.NewController(GinkgoT())
Expect(ctrl).ShouldNot(BeNil())
// Init Pipeline
- pipeline, err := queue.NewQueue(context.TODO())
- Expect(err).NotTo(HaveOccurred())
+ pipeline := queue.Local()
// Init Metadata Service
metadataService, err := metadata.NewService(context.TODO())
diff --git a/docs/api-reference.md b/docs/api-reference.md
index 956af736..537ac018 100644
--- a/docs/api-reference.md
+++ b/docs/api-reference.md
@@ -12,12 +12,6 @@
- [Catalog](#banyandb-common-v1-Catalog)
- [IntervalRule.Unit](#banyandb-common-v1-IntervalRule-Unit)
-- [banyandb/database/v1/database.proto](#banyandb_database_v1_database-proto)
- - [Node](#banyandb-database-v1-Node)
- - [Shard](#banyandb-database-v1-Shard)
-
- - [Role](#banyandb-database-v1-Role)
-
- [banyandb/model/v1/common.proto](#banyandb_model_v1_common-proto)
- [FieldValue](#banyandb-model-v1-FieldValue)
- [Float](#banyandb-model-v1-Float)
@@ -30,6 +24,30 @@
- [AggregationFunction](#banyandb-model-v1-AggregationFunction)
+- [banyandb/measure/v1/write.proto](#banyandb_measure_v1_write-proto)
+ - [DataPointValue](#banyandb-measure-v1-DataPointValue)
+ - [InternalWriteRequest](#banyandb-measure-v1-InternalWriteRequest)
+ - [WriteRequest](#banyandb-measure-v1-WriteRequest)
+ - [WriteResponse](#banyandb-measure-v1-WriteResponse)
+
+- [banyandb/stream/v1/write.proto](#banyandb_stream_v1_write-proto)
+ - [ElementValue](#banyandb-stream-v1-ElementValue)
+ - [InternalWriteRequest](#banyandb-stream-v1-InternalWriteRequest)
+ - [WriteRequest](#banyandb-stream-v1-WriteRequest)
+ - [WriteResponse](#banyandb-stream-v1-WriteResponse)
+
+- [banyandb/cluster/v1/rpc.proto](#banyandb_cluster_v1_rpc-proto)
+ - [WriteRequest](#banyandb-cluster-v1-WriteRequest)
+ - [WriteResponse](#banyandb-cluster-v1-WriteResponse)
+
+ - [Service](#banyandb-cluster-v1-Service)
+
+- [banyandb/database/v1/database.proto](#banyandb_database_v1_database-proto)
+ - [Node](#banyandb-database-v1-Node)
+ - [Shard](#banyandb-database-v1-Shard)
+
+ - [Role](#banyandb-database-v1-Role)
+
- [banyandb/model/v1/query.proto](#banyandb_model_v1_query-proto)
- [Condition](#banyandb-model-v1-Condition)
- [Criteria](#banyandb-model-v1-Criteria)
@@ -162,12 +180,6 @@
- [TopNRequest](#banyandb-measure-v1-TopNRequest)
- [TopNResponse](#banyandb-measure-v1-TopNResponse)
-- [banyandb/measure/v1/write.proto](#banyandb_measure_v1_write-proto)
- - [DataPointValue](#banyandb-measure-v1-DataPointValue)
- - [InternalWriteRequest](#banyandb-measure-v1-InternalWriteRequest)
- - [WriteRequest](#banyandb-measure-v1-WriteRequest)
- - [WriteResponse](#banyandb-measure-v1-WriteResponse)
-
- [banyandb/measure/v1/rpc.proto](#banyandb_measure_v1_rpc-proto)
- [MeasureService](#banyandb-measure-v1-MeasureService)
@@ -194,12 +206,6 @@
- [QueryRequest](#banyandb-stream-v1-QueryRequest)
- [QueryResponse](#banyandb-stream-v1-QueryResponse)
-- [banyandb/stream/v1/write.proto](#banyandb_stream_v1_write-proto)
- - [ElementValue](#banyandb-stream-v1-ElementValue)
- - [InternalWriteRequest](#banyandb-stream-v1-InternalWriteRequest)
- - [WriteRequest](#banyandb-stream-v1-WriteRequest)
- - [WriteResponse](#banyandb-stream-v1-WriteResponse)
-
- [banyandb/stream/v1/rpc.proto](#banyandb_stream_v1_rpc-proto)
- [StreamService](#banyandb-stream-v1-StreamService)
@@ -320,47 +326,136 @@ Metadata is for multi-tenant, multi-model use
-<a name="banyandb_database_v1_database-proto"></a>
+<a name="banyandb_model_v1_common-proto"></a>
<p align="right"><a href="#top">Top</a></p>
-## banyandb/database/v1/database.proto
+## banyandb/model/v1/common.proto
-<a name="banyandb-database-v1-Node"></a>
+<a name="banyandb-model-v1-FieldValue"></a>
-### Node
+### FieldValue
| Field | Type | Label | Description |
| ----- | ---- | ----- | ----------- |
-| metadata | [banyandb.common.v1.Metadata](#banyandb-common-v1-Metadata) | | |
-| roles | [Role](#banyandb-database-v1-Role) | repeated | |
-| grpc_address | [string](#string) | | |
-| http_address | [string](#string) | | |
-| created_at | [google.protobuf.Timestamp](#google-protobuf-Timestamp) | | |
+| null | [google.protobuf.NullValue](#google-protobuf-NullValue) | | |
+| str | [Str](#banyandb-model-v1-Str) | | |
+| int | [Int](#banyandb-model-v1-Int) | | |
+| binary_data | [bytes](#bytes) | | |
+| float | [Float](#banyandb-model-v1-Float) | | |
-<a name="banyandb-database-v1-Shard"></a>
+<a name="banyandb-model-v1-Float"></a>
-### Shard
+### Float
| Field | Type | Label | Description |
| ----- | ---- | ----- | ----------- |
-| id | [uint64](#uint64) | | |
-| metadata | [banyandb.common.v1.Metadata](#banyandb-common-v1-Metadata) | | |
-| catalog | [banyandb.common.v1.Catalog](#banyandb-common-v1-Catalog) | | |
-| node | [string](#string) | | |
-| total | [uint32](#uint32) | | |
-| updated_at | [google.protobuf.Timestamp](#google-protobuf-Timestamp) | | |
-| created_at | [google.protobuf.Timestamp](#google-protobuf-Timestamp) | | |
+| value | [double](#double) | | |
+
+
+
+
+
+
+<a name="banyandb-model-v1-Int"></a>
+
+### Int
+
+
+
+| Field | Type | Label | Description |
+| ----- | ---- | ----- | ----------- |
+| value | [int64](#int64) | | |
+
+
+
+
+
+
+<a name="banyandb-model-v1-IntArray"></a>
+
+### IntArray
+
+
+
+| Field | Type | Label | Description |
+| ----- | ---- | ----- | ----------- |
+| value | [int64](#int64) | repeated | |
+
+
+
+
+
+
+<a name="banyandb-model-v1-Str"></a>
+
+### Str
+
+
+
+| Field | Type | Label | Description |
+| ----- | ---- | ----- | ----------- |
+| value | [string](#string) | | |
+
+
+
+
+
+
+<a name="banyandb-model-v1-StrArray"></a>
+
+### StrArray
+
+
+
+| Field | Type | Label | Description |
+| ----- | ---- | ----- | ----------- |
+| value | [string](#string) | repeated | |
+
+
+
+
+
+
+<a name="banyandb-model-v1-TagFamilyForWrite"></a>
+
+### TagFamilyForWrite
+
+
+
+| Field | Type | Label | Description |
+| ----- | ---- | ----- | ----------- |
+| tags | [TagValue](#banyandb-model-v1-TagValue) | repeated | |
+
+
+
+
+
+
+<a name="banyandb-model-v1-TagValue"></a>
+
+### TagValue
+
+
+
+| Field | Type | Label | Description |
+| ----- | ---- | ----- | ----------- |
+| null | [google.protobuf.NullValue](#google-protobuf-NullValue) | | |
+| str | [Str](#banyandb-model-v1-Str) | | |
+| str_array | [StrArray](#banyandb-model-v1-StrArray) | | |
+| int | [Int](#banyandb-model-v1-Int) | | |
+| int_array | [IntArray](#banyandb-model-v1-IntArray) | | |
+| binary_data | [bytes](#bytes) | | |
@@ -369,18 +464,94 @@ Metadata is for multi-tenant, multi-model use
-<a name="banyandb-database-v1-Role"></a>
+<a name="banyandb-model-v1-AggregationFunction"></a>
-### Role
+### AggregationFunction
| Name | Number | Description |
| ---- | ------ | ----------- |
-| ROLE_UNSPECIFIED | 0 | |
-| ROLE_META | 1 | |
-| ROLE_DATA | 2 | |
-| ROLE_QUERY | 3 | |
-| ROLE_LIAISON | 4 | |
+| AGGREGATION_FUNCTION_UNSPECIFIED | 0 | |
+| AGGREGATION_FUNCTION_MEAN | 1 | |
+| AGGREGATION_FUNCTION_MAX | 2 | |
+| AGGREGATION_FUNCTION_MIN | 3 | |
+| AGGREGATION_FUNCTION_COUNT | 4 | |
+| AGGREGATION_FUNCTION_SUM | 5 | |
+
+
+
+
+
+
+
+
+
+
+<a name="banyandb_measure_v1_write-proto"></a>
+<p align="right"><a href="#top">Top</a></p>
+
+## banyandb/measure/v1/write.proto
+
+
+
+<a name="banyandb-measure-v1-DataPointValue"></a>
+
+### DataPointValue
+DataPointValue is the data point for writing. It only contains values.
+
+
+| Field | Type | Label | Description |
+| ----- | ---- | ----- | ----------- |
+| timestamp | [google.protobuf.Timestamp](#google-protobuf-Timestamp) | | timestamp is in the timeunit of milliseconds. |
+| tag_families | [banyandb.model.v1.TagFamilyForWrite](#banyandb-model-v1-TagFamilyForWrite) | repeated | the order of tag_families' items match the measure schema |
+| fields | [banyandb.model.v1.FieldValue](#banyandb-model-v1-FieldValue) | repeated | the order of fields match the measure schema |
+
+
+
+
+
+
+<a name="banyandb-measure-v1-InternalWriteRequest"></a>
+
+### InternalWriteRequest
+
+
+
+| Field | Type | Label | Description |
+| ----- | ---- | ----- | ----------- |
+| shard_id | [uint32](#uint32) | | |
+| series_hash | [bytes](#bytes) | | |
+| entity_values | [banyandb.model.v1.TagValue](#banyandb-model-v1-TagValue) | repeated | |
+| request | [WriteRequest](#banyandb-measure-v1-WriteRequest) | | |
+
+
+
+
+
+
+<a name="banyandb-measure-v1-WriteRequest"></a>
+
+### WriteRequest
+WriteRequest is the request contract for write
+
+
+| Field | Type | Label | Description |
+| ----- | ---- | ----- | ----------- |
+| metadata | [banyandb.common.v1.Metadata](#banyandb-common-v1-Metadata) | | the metadata is required. |
+| data_point | [DataPointValue](#banyandb-measure-v1-DataPointValue) | | the data_point is required. |
+
+
+
+
+
+
+<a name="banyandb-measure-v1-WriteResponse"></a>
+
+### WriteResponse
+WriteResponse is the response contract for write
+
+
+
@@ -389,138 +560,182 @@ Metadata is for multi-tenant, multi-model use
+
+
+
+
+<a name="banyandb_stream_v1_write-proto"></a>
+<p align="right"><a href="#top">Top</a></p>
+
+## banyandb/stream/v1/write.proto
+
+
+
+<a name="banyandb-stream-v1-ElementValue"></a>
+
+### ElementValue
+
+
+
+| Field | Type | Label | Description |
+| ----- | ---- | ----- | ----------- |
+| element_id | [string](#string) | | element_id could be span_id of a Span or segment_id of a Segment in the context of stream |
+| timestamp | [google.protobuf.Timestamp](#google-protobuf-Timestamp) | | timestamp is in the timeunit of milliseconds. It represents 1) either the start time of a Span/Segment, 2) or the timestamp of a log |
+| tag_families | [banyandb.model.v1.TagFamilyForWrite](#banyandb-model-v1-TagFamilyForWrite) | repeated | the order of tag_families' items match the stream schema |
+
+
+
+
+
+
+<a name="banyandb-stream-v1-InternalWriteRequest"></a>
+
+### InternalWriteRequest
+
+
+
+| Field | Type | Label | Description |
+| ----- | ---- | ----- | ----------- |
+| shard_id | [uint32](#uint32) | | |
+| series_hash | [bytes](#bytes) | | |
+| entity_values | [banyandb.model.v1.TagValue](#banyandb-model-v1-TagValue) | repeated | |
+| request | [WriteRequest](#banyandb-stream-v1-WriteRequest) | | |
-<a name="banyandb_model_v1_common-proto"></a>
-<p align="right"><a href="#top">Top</a></p>
-## banyandb/model/v1/common.proto
-<a name="banyandb-model-v1-FieldValue"></a>
+<a name="banyandb-stream-v1-WriteRequest"></a>
-### FieldValue
+### WriteRequest
| Field | Type | Label | Description |
| ----- | ---- | ----- | ----------- |
-| null | [google.protobuf.NullValue](#google-protobuf-NullValue) | | |
-| str | [Str](#banyandb-model-v1-Str) | | |
-| int | [Int](#banyandb-model-v1-Int) | | |
-| binary_data | [bytes](#bytes) | | |
-| float | [Float](#banyandb-model-v1-Float) | | |
-
-
-
+| metadata | [banyandb.common.v1.Metadata](#banyandb-common-v1-Metadata) | | the metadata is only required in the first write. |
+| element | [ElementValue](#banyandb-stream-v1-ElementValue) | | the element is required. |
-<a name="banyandb-model-v1-Float"></a>
-### Float
+<a name="banyandb-stream-v1-WriteResponse"></a>
-| Field | Type | Label | Description |
-| ----- | ---- | ----- | ----------- |
-| value | [double](#double) | | |
+### WriteResponse
-<a name="banyandb-model-v1-Int"></a>
+
-### Int
+
+
+
-| Field | Type | Label | Description |
-| ----- | ---- | ----- | ----------- |
-| value | [int64](#int64) | | |
+<a name="banyandb_cluster_v1_rpc-proto"></a>
+<p align="right"><a href="#top">Top</a></p>
+## banyandb/cluster/v1/rpc.proto
-<a name="banyandb-model-v1-IntArray"></a>
+<a name="banyandb-cluster-v1-WriteRequest"></a>
-### IntArray
+### WriteRequest
| Field | Type | Label | Description |
| ----- | ---- | ----- | ----------- |
-| value | [int64](#int64) | repeated | |
+| topic | [string](#string) | | |
+| message_id | [uint64](#uint64) | | |
+| stream | [banyandb.stream.v1.InternalWriteRequest](#banyandb-stream-v1-InternalWriteRequest) | | |
+| measure | [banyandb.measure.v1.InternalWriteRequest](#banyandb-measure-v1-InternalWriteRequest) | | |
-<a name="banyandb-model-v1-Str"></a>
+<a name="banyandb-cluster-v1-WriteResponse"></a>
-### Str
+### WriteResponse
-| Field | Type | Label | Description |
-| ----- | ---- | ----- | ----------- |
-| value | [string](#string) | | |
+
+
+
-<a name="banyandb-model-v1-StrArray"></a>
-### StrArray
+<a name="banyandb-cluster-v1-Service"></a>
+### Service
-| Field | Type | Label | Description |
-| ----- | ---- | ----- | ----------- |
-| value | [string](#string) | repeated | |
+| Method Name | Request Type | Response Type | Description |
+| ----------- | ------------ | ------------- | ------------|
+| Write | [WriteRequest](#banyandb-cluster-v1-WriteRequest) stream | [WriteResponse](#banyandb-cluster-v1-WriteResponse) stream | |
+
+
+<a name="banyandb_database_v1_database-proto"></a>
+<p align="right"><a href="#top">Top</a></p>
+## banyandb/database/v1/database.proto
-<a name="banyandb-model-v1-TagFamilyForWrite"></a>
-### TagFamilyForWrite
+<a name="banyandb-database-v1-Node"></a>
+
+### Node
| Field | Type | Label | Description |
| ----- | ---- | ----- | ----------- |
-| tags | [TagValue](#banyandb-model-v1-TagValue) | repeated | |
+| metadata | [banyandb.common.v1.Metadata](#banyandb-common-v1-Metadata) | | |
+| roles | [Role](#banyandb-database-v1-Role) | repeated | |
+| grpc_address | [string](#string) | | |
+| http_address | [string](#string) | | |
+| created_at | [google.protobuf.Timestamp](#google-protobuf-Timestamp) | | |
-<a name="banyandb-model-v1-TagValue"></a>
+<a name="banyandb-database-v1-Shard"></a>
-### TagValue
+### Shard
| Field | Type | Label | Description |
| ----- | ---- | ----- | ----------- |
-| null | [google.protobuf.NullValue](#google-protobuf-NullValue) | | |
-| str | [Str](#banyandb-model-v1-Str) | | |
-| str_array | [StrArray](#banyandb-model-v1-StrArray) | | |
-| int | [Int](#banyandb-model-v1-Int) | | |
-| int_array | [IntArray](#banyandb-model-v1-IntArray) | | |
-| binary_data | [bytes](#bytes) | | |
+| id | [uint64](#uint64) | | |
+| metadata | [banyandb.common.v1.Metadata](#banyandb-common-v1-Metadata) | | |
+| catalog | [banyandb.common.v1.Catalog](#banyandb-common-v1-Catalog) | | |
+| node | [string](#string) | | |
+| total | [uint32](#uint32) | | |
+| updated_at | [google.protobuf.Timestamp](#google-protobuf-Timestamp) | | |
+| created_at | [google.protobuf.Timestamp](#google-protobuf-Timestamp) | | |
@@ -529,19 +744,18 @@ Metadata is for multi-tenant, multi-model use
-<a name="banyandb-model-v1-AggregationFunction"></a>
+<a name="banyandb-database-v1-Role"></a>
-### AggregationFunction
+### Role
| Name | Number | Description |
| ---- | ------ | ----------- |
-| AGGREGATION_FUNCTION_UNSPECIFIED | 0 | |
-| AGGREGATION_FUNCTION_MEAN | 1 | |
-| AGGREGATION_FUNCTION_MAX | 2 | |
-| AGGREGATION_FUNCTION_MIN | 3 | |
-| AGGREGATION_FUNCTION_COUNT | 4 | |
-| AGGREGATION_FUNCTION_SUM | 5 | |
+| ROLE_UNSPECIFIED | 0 | |
+| ROLE_META | 1 | |
+| ROLE_DATA | 2 | |
+| ROLE_QUERY | 3 | |
+| ROLE_LIAISON | 4 | |
@@ -2433,83 +2647,6 @@ TopNResponse is the response for a query to the Query module.
-<a name="banyandb_measure_v1_write-proto"></a>
-<p align="right"><a href="#top">Top</a></p>
-
-## banyandb/measure/v1/write.proto
-
-
-
-<a name="banyandb-measure-v1-DataPointValue"></a>
-
-### DataPointValue
-DataPointValue is the data point for writing. It only contains values.
-
-
-| Field | Type | Label | Description |
-| ----- | ---- | ----- | ----------- |
-| timestamp | [google.protobuf.Timestamp](#google-protobuf-Timestamp) | | timestamp is in the timeunit of milliseconds. |
-| tag_families | [banyandb.model.v1.TagFamilyForWrite](#banyandb-model-v1-TagFamilyForWrite) | repeated | the order of tag_families' items match the measure schema |
-| fields | [banyandb.model.v1.FieldValue](#banyandb-model-v1-FieldValue) | repeated | the order of fields match the measure schema |
-
-
-
-
-
-
-<a name="banyandb-measure-v1-InternalWriteRequest"></a>
-
-### InternalWriteRequest
-
-
-
-| Field | Type | Label | Description |
-| ----- | ---- | ----- | ----------- |
-| shard_id | [uint32](#uint32) | | |
-| series_hash | [bytes](#bytes) | | |
-| entity_values | [banyandb.model.v1.TagValue](#banyandb-model-v1-TagValue) | repeated | |
-| request | [WriteRequest](#banyandb-measure-v1-WriteRequest) | | |
-
-
-
-
-
-
-<a name="banyandb-measure-v1-WriteRequest"></a>
-
-### WriteRequest
-WriteRequest is the request contract for write
-
-
-| Field | Type | Label | Description |
-| ----- | ---- | ----- | ----------- |
-| metadata | [banyandb.common.v1.Metadata](#banyandb-common-v1-Metadata) | | the metadata is required. |
-| data_point | [DataPointValue](#banyandb-measure-v1-DataPointValue) | | the data_point is required. |
-
-
-
-
-
-
-<a name="banyandb-measure-v1-WriteResponse"></a>
-
-### WriteResponse
-WriteResponse is the response contract for write
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
<a name="banyandb_measure_v1_rpc-proto"></a>
<p align="right"><a href="#top">Top</a></p>
@@ -2827,83 +2964,6 @@ QueryResponse is the response for a query to the Query module.
-<a name="banyandb_stream_v1_write-proto"></a>
-<p align="right"><a href="#top">Top</a></p>
-
-## banyandb/stream/v1/write.proto
-
-
-
-<a name="banyandb-stream-v1-ElementValue"></a>
-
-### ElementValue
-
-
-
-| Field | Type | Label | Description |
-| ----- | ---- | ----- | ----------- |
-| element_id | [string](#string) | | element_id could be span_id of a Span or segment_id of a Segment in the context of stream |
-| timestamp | [google.protobuf.Timestamp](#google-protobuf-Timestamp) | | timestamp is in the timeunit of milliseconds. It represents 1) either the start time of a Span/Segment, 2) or the timestamp of a log |
-| tag_families | [banyandb.model.v1.TagFamilyForWrite](#banyandb-model-v1-TagFamilyForWrite) | repeated | the order of tag_families' items match the stream schema |
-
-
-
-
-
-
-<a name="banyandb-stream-v1-InternalWriteRequest"></a>
-
-### InternalWriteRequest
-
-
-
-| Field | Type | Label | Description |
-| ----- | ---- | ----- | ----------- |
-| shard_id | [uint32](#uint32) | | |
-| series_hash | [bytes](#bytes) | | |
-| entity_values | [banyandb.model.v1.TagValue](#banyandb-model-v1-TagValue) | repeated | |
-| request | [WriteRequest](#banyandb-stream-v1-WriteRequest) | | |
-
-
-
-
-
-
-<a name="banyandb-stream-v1-WriteRequest"></a>
-
-### WriteRequest
-
-
-
-| Field | Type | Label | Description |
-| ----- | ---- | ----- | ----------- |
-| metadata | [banyandb.common.v1.Metadata](#banyandb-common-v1-Metadata) | | the metadata is only required in the first write. |
-| element | [ElementValue](#banyandb-stream-v1-ElementValue) | | the element is required. |
-
-
-
-
-
-
-<a name="banyandb-stream-v1-WriteResponse"></a>
-
-### WriteResponse
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
<a name="banyandb_stream_v1_rpc-proto"></a>
<p align="right"><a href="#top">Top</a></p>
diff --git a/pkg/bus/bus.go b/pkg/bus/bus.go
index 5e46c2f7..c7a8247b 100644
--- a/pkg/bus/bus.go
+++ b/pkg/bus/bus.go
@@ -45,6 +45,7 @@ type (
// Message is send on the bus to all subscribed listeners.
type Message struct {
payload payload
+ node string
id MessageID
}
@@ -58,9 +59,24 @@ func (m Message) Data() interface{} {
return m.payload
}
+// Node returns the node name of the Message.
+func (m Message) Node() string {
+ return m.node
+}
+
+// IsRemote returns true if the Message is sent from a remote node.
+func (m Message) IsRemote() bool {
+ return m.node != "local"
+}
+
// NewMessage returns a new Message with a MessageID and embed data.
func NewMessage(id MessageID, data interface{}) Message {
- return Message{id: id, payload: data}
+ return Message{id: id, node: "local", payload: data}
+}
+
+// NewMessageWithNode returns a new Message with a MessageID and NodeID and embed data.
+func NewMessageWithNode(id MessageID, node string, data interface{}) Message {
+ return Message{id: id, node: node, payload: data}
}
// MessageListener is the signature of functions that can handle an EventMessage.
@@ -103,6 +119,11 @@ func BiTopic(id string) Topic {
return Topic{id: id, typ: chTypeBidirectional}
}
+// String returns the string representation of the Topic.
+func (t Topic) String() string {
+ return t.id
+}
+
// The Bus allows publish-subscribe-style communication between components.
type Bus struct {
topics map[Topic][]channel
diff --git a/pkg/test/setup/setup.go b/pkg/test/setup/setup.go
index 2d4c957d..e7fb94c7 100644
--- a/pkg/test/setup/setup.go
+++ b/pkg/test/setup/setup.go
@@ -79,8 +79,7 @@ func CommonWithSchemaLoaders(schemaLoaders []SchemaLoader, flags ...string) (str
func modules(schemaLoaders []SchemaLoader, flags []string) func() {
// Init `Queue` module
- pipeline, err := queue.NewQueue(context.TODO())
- gomega.Expect(err).NotTo(gomega.HaveOccurred())
+ pipeline := queue.Local()
// Init `Metadata` module
metaSvc, err := metadata.NewService(context.TODO())
gomega.Expect(err).NotTo(gomega.HaveOccurred())