Merge branch 'main' into fs-architecture

apache · Aug 3, 2023 · 2731720 · 2731720
2 parents 4c21317 + 20ce7d6
commit 2731720
Show file tree

Hide file tree

Showing 40 changed files with 665 additions and 572 deletions.
diff --git a/CHANGES.md b/CHANGES.md
@@ -22,6 +22,7 @@ Release Notes.
 ### Chores
 
 - Bump several dependencies and tools.
+- Drop redundant "discovery" module from banyand. "metadata" module is enough to play the node and shard discovery role.
 
 ## 0.4.0
 

diff --git a/README.md b/README.md
@@ -29,7 +29,6 @@ The database research community usually uses [RUM conjecture](http://daslab.seas
 - [ ] Sharding
 - [ ] Load balance
 - [ ] Distributed query optimizer
-- [ ] Node discovery
 - [ ] Data queue
 
 ### Data processor

diff --git a/api/event/doc.go b/api/event/doc.go
diff --git a/api/event/measure.go b/api/event/measure.go
diff --git a/api/event/stream.go b/api/event/stream.go
diff --git a/api/proto/banyandb/database/v1/database.proto b/api/proto/banyandb/database/v1/database.proto
@@ -27,9 +27,10 @@ option java_package = "org.apache.skywalking.banyandb.database.v1";
 
 message Node {
   string id = 1;
-  string addr = 2;
-  google.protobuf.Timestamp updated_at = 3;
-  google.protobuf.Timestamp created_at = 4;
+  common.v1.Metadata metadata = 2;
+  string addr = 3;
+  google.protobuf.Timestamp updated_at = 4;
+  google.protobuf.Timestamp created_at = 5;
 }
 
 message Shard {

diff --git a/banyand/discovery/discovery.go b/banyand/discovery/discovery.go
diff --git a/banyand/internal/cmd/liaison.go b/banyand/internal/cmd/liaison.go
@@ -24,7 +24,6 @@ import (
 
 	"github.com/spf13/cobra"
 
-	"github.com/apache/skywalking-banyandb/banyand/discovery"
 	"github.com/apache/skywalking-banyandb/banyand/liaison"
 	"github.com/apache/skywalking-banyandb/banyand/liaison/http"
 	"github.com/apache/skywalking-banyandb/banyand/metadata"
@@ -42,20 +41,16 @@ var liaisonGroup = run.NewGroup("liaison")
 func newLiaisonCmd() *cobra.Command {
 	l := logger.GetLogger("bootstrap")
 	ctx := context.Background()
-	repo, err := discovery.NewServiceRepo(ctx)
-	if err != nil {
-		l.Fatal().Err(err).Msg("failed to initiate service repository")
-	}
 	// nolint: staticcheck
-	pipeline, err := queue.NewQueue(ctx, repo)
+	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")
 	}
-	tcp, err := liaison.NewEndpoint(ctx, pipeline, repo, metaSvc)
+	tcp, err := liaison.NewEndpoint(ctx, pipeline, metaSvc)
 	if err != nil {
 		l.Fatal().Err(err).Msg("failed to initiate Endpoint transport layer")
 	}
@@ -65,7 +60,6 @@ func newLiaisonCmd() *cobra.Command {
 
 	units := []run.Unit{
 		new(signal.Handler),
-		repo,
 		pipeline,
 		tcp,
 		httpServer,

diff --git a/banyand/internal/cmd/standalone.go b/banyand/internal/cmd/standalone.go
@@ -24,7 +24,6 @@ import (
 
 	"github.com/spf13/cobra"
 
-	"github.com/apache/skywalking-banyandb/banyand/discovery"
 	"github.com/apache/skywalking-banyandb/banyand/liaison"
 	"github.com/apache/skywalking-banyandb/banyand/liaison/http"
 	"github.com/apache/skywalking-banyandb/banyand/measure"
@@ -45,31 +44,27 @@ var standaloneGroup = run.NewGroup("standalone")
 func newStandaloneCmd() *cobra.Command {
 	l := logger.GetLogger("bootstrap")
 	ctx := context.Background()
-	repo, err := discovery.NewServiceRepo(ctx)
-	if err != nil {
-		l.Fatal().Err(err).Msg("failed to initiate service repository")
-	}
-	pipeline, err := queue.NewQueue(ctx, repo)
+	pipeline, err := queue.NewQueue(ctx)
 	if err != nil {
 		l.Fatal().Err(err).Msg("failed to initiate data pipeline")
 	}
 	metaSvc, err := metadata.NewService(ctx)
 	if err != nil {
 		l.Fatal().Err(err).Msg("failed to initiate metadata service")
 	}
-	streamSvc, err := stream.NewService(ctx, metaSvc, repo, pipeline)
+	streamSvc, err := stream.NewService(ctx, metaSvc, pipeline)
 	if err != nil {
 		l.Fatal().Err(err).Msg("failed to initiate stream service")
 	}
-	measureSvc, err := measure.NewService(ctx, metaSvc, repo, pipeline)
+	measureSvc, err := measure.NewService(ctx, metaSvc, pipeline)
 	if err != nil {
 		l.Fatal().Err(err).Msg("failed to initiate measure service")
 	}
 	q, err := query.NewService(ctx, streamSvc, measureSvc, metaSvc, pipeline)
 	if err != nil {
 		l.Fatal().Err(err).Msg("failed to initiate query processor")
 	}
-	tcp, err := liaison.NewEndpoint(ctx, pipeline, repo, metaSvc)
+	tcp, err := liaison.NewEndpoint(ctx, pipeline, metaSvc)
 	if err != nil {
 		l.Fatal().Err(err).Msg("failed to initiate Endpoint transport layer")
 	}
@@ -79,7 +74,6 @@ func newStandaloneCmd() *cobra.Command {
 
 	units := []run.Unit{
 		new(signal.Handler),
-		repo,
 		pipeline,
 		metaSvc,
 		measureSvc,

diff --git a/banyand/internal/cmd/storage.go b/banyand/internal/cmd/storage.go
@@ -24,7 +24,6 @@ import (
 
 	"github.com/spf13/cobra"
 
-	"github.com/apache/skywalking-banyandb/banyand/discovery"
 	"github.com/apache/skywalking-banyandb/banyand/measure"
 	"github.com/apache/skywalking-banyandb/banyand/metadata"
 	"github.com/apache/skywalking-banyandb/banyand/observability"
@@ -51,24 +50,20 @@ var flagStorageMode string
 func newStorageCmd() *cobra.Command {
 	l := logger.GetLogger("bootstrap")
 	ctx := context.Background()
-	repo, err := discovery.NewServiceRepo(ctx)
-	if err != nil {
-		l.Fatal().Err(err).Msg("failed to initiate service repository")
-	}
 	// nolint: staticcheck
-	pipeline, err := queue.NewQueue(ctx, repo)
+	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")
 	}
-	streamSvc, err := stream.NewService(ctx, metaSvc, repo, pipeline)
+	streamSvc, err := stream.NewService(ctx, metaSvc, pipeline)
 	if err != nil {
 		l.Fatal().Err(err).Msg("failed to initiate stream service")
 	}
-	measureSvc, err := measure.NewService(ctx, metaSvc, repo, pipeline)
+	measureSvc, err := measure.NewService(ctx, metaSvc, pipeline)
 	if err != nil {
 		l.Fatal().Err(err).Msg("failed to initiate measure service")
 	}
@@ -82,7 +77,6 @@ func newStorageCmd() *cobra.Command {
 
 	units := []run.Unit{
 		new(signal.Handler),
-		repo,
 		pipeline,
 		measureSvc,
 		streamSvc,