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 22:04:55 UTC
[skywalking-banyandb] branch queue created (now ebeab1fe)
This is an automated email from the ASF dual-hosted git repository.
hanahmily pushed a change to branch queue
in repository https://gitbox.apache.org/repos/asf/skywalking-banyandb.git
at ebeab1fe Implement remote queue
This branch includes the following new commits:
new ebeab1fe Implement remote queue
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
[skywalking-banyandb] 01/01: Implement remote queue
Posted by ha...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
hanahmily pushed a commit to branch queue
in repository https://gitbox.apache.org/repos/asf/skywalking-banyandb.git
commit ebeab1fe6f25e4a6cc9e3839c44e3e70f22c095f
Author: Gao Hongtao <ha...@gmail.com>
AuthorDate: Mon Aug 28 13:37:19 2023 +0000
Implement remote queue
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 | 2 +-
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, 1004 insertions(+), 338 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..0873c801 100644
--- a/banyand/liaison/grpc/stream.go
+++ b/banyand/liaison/grpc/stream.go
@@ -101,7 +101,7 @@ 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)
+ message := bus.NewMessageWithNode(bus.MessageID(time.Now().UnixNano()), "TODO", iwr)
_, errWritePub := s.pipeline.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())