You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@skywalking.apache.org by GitBox <gi...@apache.org> on 2021/08/07 02:24:02 UTC
[GitHub] [skywalking-banyandb] Fine0830 opened a new pull request #30: feat: Implement Write in the liaison
Fine0830 opened a new pull request #30:
URL: https://github.com/apache/skywalking-banyandb/pull/30
Implement Write in the liaison
Signed-off-by: Qiuxia Fan <qi...@apache.org>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@skywalking.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [skywalking-banyandb] Fine0830 commented on a change in pull request #30: feat: Implement Write in the liaison
Posted by GitBox <gi...@apache.org>.
Fine0830 commented on a change in pull request #30:
URL: https://github.com/apache/skywalking-banyandb/pull/30#discussion_r685795244
##########
File path: banyand/liaison/grpc/grpc.go
##########
@@ -49,26 +49,35 @@ import (
)
var (
- ErrSeriesEvents = errors.New("no seriesEvent")
- ErrShardEvents = errors.New("no shardEvent")
- ErrInvalidSeriesID = errors.New("invalid seriesID")
+ ErrSeriesEvents = errors.New("no seriesEvent")
+ ErrShardEvents = errors.New("no shardEvent")
+ ErrInvalidSeriesID = errors.New("invalid seriesID")
+ ErrServerCert = errors.New("invalid server cert file")
+ ErrServerKey = errors.New("invalid server key file")
+ ErrServerHostOverride = errors.New("invalid serverHostOverride")
+ ErrNoAddr = errors.New("no address")
)
type Server struct {
- addr string
- log *logger.Logger
- ser *grpclib.Server
- pipeline queue.Queue
- repo discovery.ServiceRepo
- shardInfo *shardInfo
- seriesInfo *seriesInfo
+ addr string
+ maxRecvMsgSize int
+ TLSVal bool
+ ServerHostOverride string
+ CertFile string
Review comment:
I export them for the `grpc_test`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@skywalking.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [skywalking-banyandb] hanahmily merged pull request #30: feat: Implement Write in the liaison
Posted by GitBox <gi...@apache.org>.
hanahmily merged pull request #30:
URL: https://github.com/apache/skywalking-banyandb/pull/30
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@skywalking.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [skywalking-banyandb] hanahmily commented on a change in pull request #30: feat: Implement Write in the liaison
Posted by GitBox <gi...@apache.org>.
hanahmily commented on a change in pull request #30:
URL: https://github.com/apache/skywalking-banyandb/pull/30#discussion_r684636138
##########
File path: banyand/liaison/grpc/grpc.go
##########
@@ -116,10 +142,25 @@ func (s *Server) Serve() error {
if err != nil {
s.log.Fatal().Err(err).Msg("Failed to listen")
}
-
- s.ser = grpc.NewServer()
- // TODO: add server implementation here
- v1.RegisterTraceServiceServer(s.ser, v1.UnimplementedTraceServiceServer{})
+ var opts []grpclib.ServerOption
+ var f embed.FS
+ if *tls {
+ if *certFile == "" {
+ serverCert, _ := f.ReadFile("data/x509/server_cert.pem")
+ *certFile = string(serverCert)
+ }
+ if *keyFile == "" {
+ serverKey, _ := f.ReadFile("data/x509/server_key.pem")
+ *keyFile = string(serverKey)
+ }
+ cred, err := credentials.NewServerTLSFromFile(*certFile, *keyFile)
+ if err != nil {
+ log.Fatalf("Failed to generate credentials %v", err)
+ }
+ opts = []grpclib.ServerOption{grpclib.Creds(cred)}
+ }
+ s.ser = grpclib.NewServer(opts...)
Review comment:
we should create a flag to set [MaxRecvMsgSize](https://github.com/grpc/grpc-go/blob/master/server.go#L347) in case binary data is more than the default size.
##########
File path: banyand/liaison/data/x509/server_cert.pem
##########
@@ -0,0 +1,32 @@
+-----BEGIN CERTIFICATE-----
Review comment:
This file is for test purposes? If that, pls move them to `liaison/testdata/...`
##########
File path: banyand/liaison/grpc/grpc.go
##########
@@ -128,3 +169,87 @@ func (s *Server) GracefulStop() {
s.log.Info().Msg("stopping")
s.ser.GracefulStop()
}
+
+func assemblyWriteData(shardID uint, writeEntity *v1.WriteRequest, seriesID uint64) data.TraceWriteDate {
+ return data.TraceWriteDate{ShardID: shardID, SeriesID: seriesID, WriteRequest: writeEntity}
+}
+
+func (s *Server) Write(TraceWriteServer v1.TraceService_WriteServer) error {
+ for {
+ writeEntity, err := TraceWriteServer.Recv()
+ if err == io.EOF {
+ return nil
+ }
+ if err != nil {
+ return err
+ }
+
+ //log.Println("writeEntity:", writeEntity)
+ ana := logical.DefaultAnalyzer()
+ metadata := common.Metadata{
+ KindVersion: apischema.SeriesKindVersion,
+ Spec: writeEntity.GetMetadata(),
+ }
+ schema, ruleError := ana.BuildTraceSchema(context.TODO(), metadata)
+ if ruleError != nil {
+ return ruleError
+ }
+ if s.seriesInfo.seriesEvent == nil {
Review comment:
Data race here. Pls leverage `sync.RWMutex` to protect seriesEvent.
##########
File path: banyand/liaison/grpc/grpc.go
##########
@@ -116,10 +142,25 @@ func (s *Server) Serve() error {
if err != nil {
s.log.Fatal().Err(err).Msg("Failed to listen")
}
-
- s.ser = grpc.NewServer()
- // TODO: add server implementation here
- v1.RegisterTraceServiceServer(s.ser, v1.UnimplementedTraceServiceServer{})
+ var opts []grpclib.ServerOption
+ var f embed.FS
+ if *tls {
+ if *certFile == "" {
+ serverCert, _ := f.ReadFile("data/x509/server_cert.pem")
Review comment:
Why would you set default cert and key files here? Once they are absent with tls=true, you should return an error in https://github.com/apache/skywalking-banyandb/blob/199c1b0334b8cf4851fe1f1e8b0ebd2bf3f665db/banyand/liaison/grpc/grpc.go#L110
##########
File path: banyand/liaison/grpc/grpc_test.go
##########
@@ -0,0 +1,203 @@
+// 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 grpc_test
+
+import (
+ "context"
+ "io"
+ "log"
+ "os"
+ "path"
+ "testing"
+ "time"
+
+ googleUUID "github.com/google/uuid"
+ "github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
+ grpclib "google.golang.org/grpc"
+
+ v1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/v1"
+ "github.com/apache/skywalking-banyandb/banyand/discovery"
+ "github.com/apache/skywalking-banyandb/banyand/index"
+ "github.com/apache/skywalking-banyandb/banyand/liaison"
+ "github.com/apache/skywalking-banyandb/banyand/query"
+ "github.com/apache/skywalking-banyandb/banyand/queue"
+ "github.com/apache/skywalking-banyandb/banyand/series/trace"
+ "github.com/apache/skywalking-banyandb/banyand/storage"
+ "github.com/apache/skywalking-banyandb/pkg/logger"
+ "github.com/apache/skywalking-banyandb/pkg/pb"
+)
+
+var (
+ serverAddr = "localhost:17912"
+)
+
+func setupService(t *testing.T, tester *require.Assertions) func() {
+ tester.NoError(logger.Init(logger.Logging{
+ Env: "dev",
+ Level: "warn",
+ }))
+ // Init `Discovery` module
+ repo, err := discovery.NewServiceRepo(context.Background())
+ tester.NoError(err)
+ tester.NotNil(repo)
+ // Init `Queue` module
+ pipeline, err := queue.NewQueue(context.TODO(), repo)
+ tester.NoError(err)
+ // Init `Database` module
+ db, err := storage.NewDB(context.TODO(), repo)
+ tester.NoError(err)
+ uuid, err := googleUUID.NewUUID()
+ tester.NoError(err)
+ rootPath := path.Join(os.TempDir(), "banyandb-"+uuid.String())
+ tester.NoError(db.FlagSet().Parse([]string{"--root-path=" + rootPath}))
+ // Init `Index` module
+ indexSvc, err := index.NewService(context.TODO(), repo)
+ tester.NoError(err)
+ // Init `Trace` module
+ traceSvc, err := trace.NewService(context.TODO(), db, repo, indexSvc, pipeline)
+ tester.NoError(err)
+ // Init `Query` module
+ executor, err := query.NewExecutor(context.TODO(), repo, indexSvc, traceSvc, traceSvc)
+ tester.NoError(err)
+ // Init `liaison` module
+ tcp, err := liaison.NewEndpoint(context.TODO(), pipeline, repo)
+ tester.NoError(err)
+
+ err = indexSvc.PreRun()
+ tester.NoError(err)
+
+ err = traceSvc.PreRun()
+ tester.NoError(err)
+
+ err = db.PreRun()
+ tester.NoError(err)
+
+ err = executor.PreRun()
+ tester.NoError(err)
+
+ tcp.FlagSet()
+ err = tcp.PreRun()
+ tester.NoError(err)
+
+ go func() {
+ tester.NoError(traceSvc.Serve())
+ tester.NoError(err)
+ }()
+
+ go func() {
+ tester.NoError(tcp.Serve())
+ tester.NoError(err)
+ }()
+
+ ctx, cancelFunc := context.WithTimeout(context.Background(), 10*time.Second)
+ defer cancelFunc()
+
+ tester.True(indexSvc.Ready(ctx, index.MetaExists("default", "sw")))
+
+ return func() {
+ db.GracefulStop()
+ _ = os.RemoveAll(rootPath)
+ }
+}
+
+func TestTraceWrite(t *testing.T) {
+ tester := require.New(t)
+ gracefulStop := setupService(t, tester)
+ defer gracefulStop()
+
+ conn, err := grpclib.Dial(serverAddr, grpclib.WithInsecure())
+ assert.NoError(t, err)
+ defer conn.Close()
+
+ client := v1.NewTraceServiceClient(conn)
+ ctx := context.Background()
+ entityValue := pb.NewEntityValueBuilder().
+ EntityID("entityId").
+ DataBinary([]byte{12}).
+ Fields("trace_id-xxfff.111323",
+ 0,
+ "webapp_id",
+ "10.0.0.1_id",
+ "/home_id",
+ "webapp",
+ "10.0.0.1",
+ "/home",
+ 300,
+ 1622933202000000000).
+ Timestamp(time.Now()).
+ Build()
+ criteria := pb.NewWriteEntityBuilder().
+ EntityValue(entityValue).
+ Metadata("default", "sw").
+ Build()
+ stream, errorWrite := client.Write(ctx)
+ if errorWrite != nil {
+ log.Fatalf("%v.runWrite(_) = _, %v", client, errorWrite)
+ }
+ waitc := make(chan struct{})
+ go func() {
+ for {
+ writeResponse, errRecv := stream.Recv()
+ if errRecv == io.EOF {
+ // read done.
+ close(waitc)
+ return
+ }
+ if errRecv != nil {
+ log.Fatalf("Failed to receive data : %v", errRecv)
+ }
+ log.Println("writeResponse: ", writeResponse)
Review comment:
to user `assert` to replace `log`
##########
File path: banyand/series/trace/service.go
##########
@@ -65,7 +72,9 @@ func (s *service) PreRun() error {
return err
}
s.schemaMap = make(map[string]*traceSeries, len(schemas))
+ s.writeListener.schemaMap = make(map[string]*traceSeries, len(schemas))
s.l = logger.GetLogger(s.Name())
+ s.writeListener.l = logger.GetLogger(s.Name())
Review comment:
pls assign `s.l` to `s.writeListener.l`
##########
File path: banyand/liaison/grpc/grpc.go
##########
@@ -128,3 +169,87 @@ func (s *Server) GracefulStop() {
s.log.Info().Msg("stopping")
s.ser.GracefulStop()
}
+
+func assemblyWriteData(shardID uint, writeEntity *v1.WriteRequest, seriesID uint64) data.TraceWriteDate {
+ return data.TraceWriteDate{ShardID: shardID, SeriesID: seriesID, WriteRequest: writeEntity}
+}
+
+func (s *Server) Write(TraceWriteServer v1.TraceService_WriteServer) error {
+ for {
+ writeEntity, err := TraceWriteServer.Recv()
+ if err == io.EOF {
+ return nil
+ }
+ if err != nil {
+ return err
+ }
+
+ //log.Println("writeEntity:", writeEntity)
+ ana := logical.DefaultAnalyzer()
+ metadata := common.Metadata{
+ KindVersion: apischema.SeriesKindVersion,
+ Spec: writeEntity.GetMetadata(),
+ }
+ schema, ruleError := ana.BuildTraceSchema(context.TODO(), metadata)
+ if ruleError != nil {
+ return ruleError
+ }
+ if s.seriesInfo.seriesEvent == nil {
+ return errors.New("No seriesEvents")
Review comment:
Could you create a global error and return it here?
##########
File path: banyand/liaison/grpc/grpc.go
##########
@@ -128,3 +169,87 @@ func (s *Server) GracefulStop() {
s.log.Info().Msg("stopping")
s.ser.GracefulStop()
}
+
+func assemblyWriteData(shardID uint, writeEntity *v1.WriteRequest, seriesID uint64) data.TraceWriteDate {
+ return data.TraceWriteDate{ShardID: shardID, SeriesID: seriesID, WriteRequest: writeEntity}
+}
+
+func (s *Server) Write(TraceWriteServer v1.TraceService_WriteServer) error {
+ for {
+ writeEntity, err := TraceWriteServer.Recv()
+ if err == io.EOF {
+ return nil
+ }
+ if err != nil {
+ return err
+ }
+
+ //log.Println("writeEntity:", writeEntity)
Review comment:
pls remove debugging codes.
##########
File path: banyand/series/trace/service.go
##########
@@ -65,7 +72,9 @@ func (s *service) PreRun() error {
return err
}
s.schemaMap = make(map[string]*traceSeries, len(schemas))
+ s.writeListener.schemaMap = make(map[string]*traceSeries, len(schemas))
Review comment:
Could you assign `s.schemaMap` to `s.writeListener.schemaMap` directly?
##########
File path: banyand/liaison/grpc/grpc.go
##########
@@ -19,31 +19,54 @@ package grpc
import (
"context"
+ "flag"
+ "fmt"
+ "io"
+ "log"
"net"
+ "strings"
+ "time"
- "google.golang.org/grpc"
+ "github.com/pkg/errors"
+ grpclib "google.golang.org/grpc"
+ "google.golang.org/grpc/credentials"
+ "github.com/apache/skywalking-banyandb/api/common"
+ "github.com/apache/skywalking-banyandb/api/data"
"github.com/apache/skywalking-banyandb/api/event"
v1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/v1"
+ apischema "github.com/apache/skywalking-banyandb/api/schema"
"github.com/apache/skywalking-banyandb/banyand/discovery"
+ serverData "github.com/apache/skywalking-banyandb/banyand/liaison/data"
"github.com/apache/skywalking-banyandb/banyand/queue"
"github.com/apache/skywalking-banyandb/pkg/bus"
+ "github.com/apache/skywalking-banyandb/pkg/convert"
"github.com/apache/skywalking-banyandb/pkg/logger"
+ "github.com/apache/skywalking-banyandb/pkg/partition"
+ "github.com/apache/skywalking-banyandb/pkg/query/logical"
"github.com/apache/skywalking-banyandb/pkg/run"
)
+var (
+ tls = flag.Bool("tls", false, "Connection uses TLS if true, else plain TCP")
Review comment:
Exactly. move these flags to the above function by referring to https://github.com/apache/skywalking-banyandb/blob/199c1b0334b8cf4851fe1f1e8b0ebd2bf3f665db/banyand/liaison/grpc/grpc.go#L104
##########
File path: banyand/liaison/grpc/grpc_test.go
##########
@@ -0,0 +1,203 @@
+// 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 grpc_test
+
+import (
+ "context"
+ "io"
+ "log"
+ "os"
+ "path"
+ "testing"
+ "time"
+
+ googleUUID "github.com/google/uuid"
+ "github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
+ grpclib "google.golang.org/grpc"
+
+ v1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/v1"
+ "github.com/apache/skywalking-banyandb/banyand/discovery"
+ "github.com/apache/skywalking-banyandb/banyand/index"
+ "github.com/apache/skywalking-banyandb/banyand/liaison"
+ "github.com/apache/skywalking-banyandb/banyand/query"
+ "github.com/apache/skywalking-banyandb/banyand/queue"
+ "github.com/apache/skywalking-banyandb/banyand/series/trace"
+ "github.com/apache/skywalking-banyandb/banyand/storage"
+ "github.com/apache/skywalking-banyandb/pkg/logger"
+ "github.com/apache/skywalking-banyandb/pkg/pb"
+)
+
+var (
+ serverAddr = "localhost:17912"
+)
+
+func setupService(t *testing.T, tester *require.Assertions) func() {
+ tester.NoError(logger.Init(logger.Logging{
+ Env: "dev",
+ Level: "warn",
+ }))
+ // Init `Discovery` module
+ repo, err := discovery.NewServiceRepo(context.Background())
+ tester.NoError(err)
+ tester.NotNil(repo)
+ // Init `Queue` module
+ pipeline, err := queue.NewQueue(context.TODO(), repo)
+ tester.NoError(err)
+ // Init `Database` module
+ db, err := storage.NewDB(context.TODO(), repo)
+ tester.NoError(err)
+ uuid, err := googleUUID.NewUUID()
+ tester.NoError(err)
+ rootPath := path.Join(os.TempDir(), "banyandb-"+uuid.String())
+ tester.NoError(db.FlagSet().Parse([]string{"--root-path=" + rootPath}))
+ // Init `Index` module
+ indexSvc, err := index.NewService(context.TODO(), repo)
+ tester.NoError(err)
+ // Init `Trace` module
+ traceSvc, err := trace.NewService(context.TODO(), db, repo, indexSvc, pipeline)
+ tester.NoError(err)
+ // Init `Query` module
+ executor, err := query.NewExecutor(context.TODO(), repo, indexSvc, traceSvc, traceSvc)
+ tester.NoError(err)
+ // Init `liaison` module
+ tcp, err := liaison.NewEndpoint(context.TODO(), pipeline, repo)
+ tester.NoError(err)
+
+ err = indexSvc.PreRun()
+ tester.NoError(err)
+
+ err = traceSvc.PreRun()
+ tester.NoError(err)
+
+ err = db.PreRun()
+ tester.NoError(err)
+
+ err = executor.PreRun()
+ tester.NoError(err)
+
+ tcp.FlagSet()
+ err = tcp.PreRun()
+ tester.NoError(err)
+
+ go func() {
+ tester.NoError(traceSvc.Serve())
+ tester.NoError(err)
+ }()
+
+ go func() {
+ tester.NoError(tcp.Serve())
+ tester.NoError(err)
+ }()
+
+ ctx, cancelFunc := context.WithTimeout(context.Background(), 10*time.Second)
+ defer cancelFunc()
+
+ tester.True(indexSvc.Ready(ctx, index.MetaExists("default", "sw")))
+
+ return func() {
+ db.GracefulStop()
+ _ = os.RemoveAll(rootPath)
+ }
+}
+
+func TestTraceWrite(t *testing.T) {
+ tester := require.New(t)
+ gracefulStop := setupService(t, tester)
+ defer gracefulStop()
+
+ conn, err := grpclib.Dial(serverAddr, grpclib.WithInsecure())
+ assert.NoError(t, err)
+ defer conn.Close()
+
+ client := v1.NewTraceServiceClient(conn)
+ ctx := context.Background()
+ entityValue := pb.NewEntityValueBuilder().
+ EntityID("entityId").
+ DataBinary([]byte{12}).
+ Fields("trace_id-xxfff.111323",
+ 0,
+ "webapp_id",
+ "10.0.0.1_id",
+ "/home_id",
+ "webapp",
+ "10.0.0.1",
+ "/home",
+ 300,
+ 1622933202000000000).
+ Timestamp(time.Now()).
+ Build()
+ criteria := pb.NewWriteEntityBuilder().
+ EntityValue(entityValue).
+ Metadata("default", "sw").
+ Build()
+ stream, errorWrite := client.Write(ctx)
+ if errorWrite != nil {
+ log.Fatalf("%v.runWrite(_) = _, %v", client, errorWrite)
+ }
+ waitc := make(chan struct{})
+ go func() {
+ for {
+ writeResponse, errRecv := stream.Recv()
+ if errRecv == io.EOF {
+ // read done.
+ close(waitc)
+ return
+ }
+ if errRecv != nil {
+ log.Fatalf("Failed to receive data : %v", errRecv)
Review comment:
to use `assert.NoError`(testify) here
##########
File path: banyand/liaison/grpc/grpc.go
##########
@@ -128,3 +169,87 @@ func (s *Server) GracefulStop() {
s.log.Info().Msg("stopping")
s.ser.GracefulStop()
}
+
+func assemblyWriteData(shardID uint, writeEntity *v1.WriteRequest, seriesID uint64) data.TraceWriteDate {
+ return data.TraceWriteDate{ShardID: shardID, SeriesID: seriesID, WriteRequest: writeEntity}
+}
+
+func (s *Server) Write(TraceWriteServer v1.TraceService_WriteServer) error {
+ for {
+ writeEntity, err := TraceWriteServer.Recv()
+ if err == io.EOF {
+ return nil
+ }
+ if err != nil {
+ return err
+ }
+
+ //log.Println("writeEntity:", writeEntity)
+ ana := logical.DefaultAnalyzer()
+ metadata := common.Metadata{
+ KindVersion: apischema.SeriesKindVersion,
+ Spec: writeEntity.GetMetadata(),
+ }
+ schema, ruleError := ana.BuildTraceSchema(context.TODO(), metadata)
+ if ruleError != nil {
+ return ruleError
+ }
+ if s.seriesInfo.seriesEvent == nil {
+ return errors.New("No seriesEvents")
+ }
+ seriesIDLen := len(s.seriesInfo.seriesEvent.FieldNamesCompositeSeriesId)
+ var str string
+ var arr []string
+ for i := 0; i < seriesIDLen; i++ {
+ id := s.seriesInfo.seriesEvent.FieldNamesCompositeSeriesId[i]
+ if defined, sub := schema.FieldSubscript(id); defined {
+ for _, field := range writeEntity.GetEntity().GetFields() {
+ switch v := field.GetValueType().(type) {
+ case *v1.Field_StrArray:
+ for j := 0; j < len(v.StrArray.Value); j++ {
+ if sub == j {
+ arr = append(arr, v.StrArray.Value[j])
+ }
+ }
+ case *v1.Field_IntArray:
+ for t := 0; t < len(v.IntArray.Value); t++ {
+ arr = append(arr, fmt.Sprint(v.IntArray.Value[t]))
+ }
+ case *v1.Field_Int:
+ arr = append(arr, fmt.Sprint(v.Int.Value))
+ case *v1.Field_Str:
+ arr = append(arr, fmt.Sprint(v.Str.Value))
+ }
+ }
+ }
+ }
+ str = strings.Join(arr, "")
+ if str == "" {
+ return errors.New("invalid seriesID")
+ }
+ seriesID := []byte(str)
+ shardNum := s.shardInfo.shardEvent.GetShard().GetId()
Review comment:
shardEvent might be nil, pls check whether it's nil before read it. and `sync.RWMutex` is also needed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@skywalking.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [skywalking-banyandb] hanahmily commented on a change in pull request #30: feat: Implement Write in the liaison
Posted by GitBox <gi...@apache.org>.
hanahmily commented on a change in pull request #30:
URL: https://github.com/apache/skywalking-banyandb/pull/30#discussion_r685573347
##########
File path: banyand/liaison/grpc/grpc.go
##########
@@ -102,12 +135,50 @@ func (s *Server) Name() string {
}
func (s *Server) FlagSet() *run.FlagSet {
+ size := 1024 * 1024 * 8
+ _, currentFile, _, _ := runtime.Caller(0)
+ basePath := filepath.Dir(currentFile)
+ serverCert := filepath.Join(basePath, "testdata/server_cert.pem")
+ serverKey := filepath.Join(basePath, "testdata/server_key.pem")
+
fs := run.NewFlagSet("grpc")
- fs.StringVarP(&s.addr, "addr", "", ":17912", "the address of banyand listens")
+ fs.Int("maxRecvMsgSize", size, "The size of max receiving message")
Review comment:
Pls follow `addr` to set up flags:
```
fs.StringVarP(&s.addr, "addr", "", ":17912", "The address of banyand listens")
```
##########
File path: banyand/liaison/grpc/testdata/server_cert.pem
##########
@@ -0,0 +1,32 @@
+-----BEGIN CERTIFICATE-----
Review comment:
As @kezhenxu94 mentioned, if you copy https://github.com/grpc/grpc-go/blob/master/examples/data/x509/server_cert.pem, pls add notices and license in banyandb. But they look like only for the test purpose, how about generate them by your hand
##########
File path: banyand/liaison/grpc/grpc.go
##########
@@ -52,6 +77,9 @@ func (s *shardInfo) Rev(message bus.Message) (resp bus.Message) {
s.log.Warn().Msg("invalid event data type")
return
}
+ s.RWMutex.Lock()
+ defer s.RWMutex.Unlock()
+ s.shardEvent = shardEvent
s.log.Info().
Review comment:
* Add two separate `mutex` to protect `shardEvent` and `seriesEvent`. They don't want to block each other.
* Pls parse the `action` of `shardEvent` and `seriesEvent` instead of accepting it directly.
##########
File path: banyand/liaison/grpc/grpc.go
##########
@@ -128,3 +210,92 @@ func (s *Server) GracefulStop() {
s.log.Info().Msg("stopping")
s.ser.GracefulStop()
}
+
+func assemblyWriteData(shardID uint, writeEntity *v1.WriteRequest, seriesID uint64) data.TraceWriteDate {
+ return data.TraceWriteDate{ShardID: shardID, SeriesID: seriesID, WriteRequest: writeEntity}
+}
+
+func (s *Server) Write(TraceWriteServer v1.TraceService_WriteServer) error {
+ for {
+ writeEntity, err := TraceWriteServer.Recv()
+ if err == io.EOF {
+ return nil
+ }
+ if err != nil {
+ return err
+ }
+
+ ana := logical.DefaultAnalyzer()
+ metadata := common.Metadata{
+ KindVersion: apischema.SeriesKindVersion,
+ Spec: writeEntity.GetMetadata(),
+ }
+ schema, ruleError := ana.BuildTraceSchema(context.TODO(), metadata)
+ if ruleError != nil {
+ return ruleError
+ }
+ s.seriesInfo.RWMutex.RLock()
+ if s.seriesInfo.seriesEvent == nil {
+ s.seriesInfo.RWMutex.RUnlock()
Review comment:
How about adding a `s.getSeriesInfo` here like
```go
func (s *Server) getSeriesInfo() {
s.RWMutex.RLock()
defer s.RWMutext.RUlock()
return s.xxxx
}
```
The above method helps us get rid of checking where to unlock rwmutext.
##########
File path: banyand/liaison/grpc/grpc.go
##########
@@ -102,12 +135,50 @@ func (s *Server) Name() string {
}
func (s *Server) FlagSet() *run.FlagSet {
+ size := 1024 * 1024 * 8
+ _, currentFile, _, _ := runtime.Caller(0)
+ basePath := filepath.Dir(currentFile)
+ serverCert := filepath.Join(basePath, "testdata/server_cert.pem")
+ serverKey := filepath.Join(basePath, "testdata/server_key.pem")
+
fs := run.NewFlagSet("grpc")
- fs.StringVarP(&s.addr, "addr", "", ":17912", "the address of banyand listens")
+ fs.Int("maxRecvMsgSize", size, "The size of max receiving message")
+ fs.Bool("tls", true, "Connection uses TLS if true, else plain TCP")
+ fs.String("certFile", serverCert, "The TLS cert file")
+ fs.String("keyFile", serverKey, "The TLS key file")
+ fs.String("serverHostOverride", "x.test.example.com", "The server name used to verify the hostname returned by the TLS handshake")
+ fs.StringVarP(&s.addr, "addr", "", ":17912", "The address of banyand listens")
+
return fs
}
func (s *Server) Validate() error {
+ _, err := s.FlagSet().GetInt("maxRecvMsgSize")
Review comment:
You could NOT access s.FlagSet() which will be called by module lifecycle framework.
Let's use `addr` to show how to validate flags:
```go
func (s *Server) Validate() error {
if s.addr == "" {
return errNoAddr
}
}
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@skywalking.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [skywalking-banyandb] hanahmily commented on a change in pull request #30: feat: Implement Write in the liaison
Posted by GitBox <gi...@apache.org>.
hanahmily commented on a change in pull request #30:
URL: https://github.com/apache/skywalking-banyandb/pull/30#discussion_r685746426
##########
File path: banyand/liaison/grpc/testdata/server_cert.pem
##########
@@ -0,0 +1,32 @@
+-----BEGIN CERTIFICATE-----
Review comment:
To copy source codes needs another process more than lib dependency. It's much easy to generate TLS files by referring to https://bbengfort.github.io/2017/03/secure-grpc/
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@skywalking.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [skywalking-banyandb] kezhenxu94 commented on pull request #30: feat: Implement Write in the liaison
Posted by GitBox <gi...@apache.org>.
kezhenxu94 commented on pull request #30:
URL: https://github.com/apache/skywalking-banyandb/pull/30#issuecomment-894595131
Also, as we are using Go 1.16 I'm wondering whether it's necessary to copy `banyand/liaison/data/data.go`, maybe we can use built-in [go:embed](https://golang.org/pkg/embed/) since Go 1.16
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@skywalking.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [skywalking-banyandb] lujiajing1126 commented on pull request #30: feat: Implement Write in the liaison
Posted by GitBox <gi...@apache.org>.
lujiajing1126 commented on pull request #30:
URL: https://github.com/apache/skywalking-banyandb/pull/30#issuecomment-895192053
LGTM!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@skywalking.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [skywalking-banyandb] Fine0830 commented on a change in pull request #30: feat: Implement Write in the liaison
Posted by GitBox <gi...@apache.org>.
Fine0830 commented on a change in pull request #30:
URL: https://github.com/apache/skywalking-banyandb/pull/30#discussion_r685795465
##########
File path: banyand/liaison/grpc/testdata/server_cert.pem
##########
@@ -0,0 +1,32 @@
+-----BEGIN CERTIFICATE-----
Review comment:
OK
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@skywalking.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [skywalking-banyandb] lujiajing1126 commented on a change in pull request #30: feat: Implement Write in the liaison
Posted by GitBox <gi...@apache.org>.
lujiajing1126 commented on a change in pull request #30:
URL: https://github.com/apache/skywalking-banyandb/pull/30#discussion_r684577263
##########
File path: banyand/liaison/grpc/grpc.go
##########
@@ -128,3 +166,87 @@ func (s *Server) GracefulStop() {
s.log.Info().Msg("stopping")
s.ser.GracefulStop()
}
+
+func assemblyWriteData(shardID uint, writeEntity *v1.WriteRequest, seriesID uint64) data.TraceWriteDate {
+ return data.TraceWriteDate{ShardID: shardID, SeriesID: seriesID, WriteRequest: writeEntity}
+}
+
+func (s *Server) Write(TraceWriteServer v1.TraceService_WriteServer) error {
+ for {
+ writeEntity, err := TraceWriteServer.Recv()
+ if err == io.EOF {
+ return nil
+ }
+ if err != nil {
+ return err
+ }
+
+ //log.Println("writeEntity:", writeEntity)
+ ana := logical.DefaultAnalyzer()
+ metadata := common.Metadata{
+ KindVersion: apischema.SeriesKindVersion,
+ Spec: writeEntity.GetMetadata(),
+ }
+ schema, ruleError := ana.BuildTraceSchema(context.TODO(), metadata)
+ if ruleError != nil {
+ return ruleError
+ }
+ if s.seriesInfo.seriesEvent == nil {
+ return errors.New("No seriesEvents")
+ }
+ seriesIDLen := len(s.seriesInfo.seriesEvent.FieldNamesCompositeSeriesId)
+ var str string
+ var arr []string
+ for i := 0; i < seriesIDLen; i++ {
Review comment:
I suggest optimizing this loop by,
1) first converting `FieldNamesCompositeSeriesId` to `[]*FieldRef` (i.e. `FieldSubscript` in your case, maybe you can make use of `Schema.CreateRef(names ...string)` to do this),
2) then iterating over fields (`writeEntity.GetEntity().GetFields()`) to compose the seriesID,
If we do so, only one iteration is necessary.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@skywalking.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [skywalking-banyandb] Fine0830 commented on a change in pull request #30: feat: Implement Write in the liaison
Posted by GitBox <gi...@apache.org>.
Fine0830 commented on a change in pull request #30:
URL: https://github.com/apache/skywalking-banyandb/pull/30#discussion_r684606363
##########
File path: banyand/liaison/grpc/grpc.go
##########
@@ -128,3 +166,87 @@ func (s *Server) GracefulStop() {
s.log.Info().Msg("stopping")
s.ser.GracefulStop()
}
+
+func assemblyWriteData(shardID uint, writeEntity *v1.WriteRequest, seriesID uint64) data.TraceWriteDate {
+ return data.TraceWriteDate{ShardID: shardID, SeriesID: seriesID, WriteRequest: writeEntity}
+}
+
+func (s *Server) Write(TraceWriteServer v1.TraceService_WriteServer) error {
+ for {
+ writeEntity, err := TraceWriteServer.Recv()
+ if err == io.EOF {
+ return nil
+ }
+ if err != nil {
+ return err
+ }
+
+ //log.Println("writeEntity:", writeEntity)
+ ana := logical.DefaultAnalyzer()
+ metadata := common.Metadata{
+ KindVersion: apischema.SeriesKindVersion,
+ Spec: writeEntity.GetMetadata(),
+ }
+ schema, ruleError := ana.BuildTraceSchema(context.TODO(), metadata)
+ if ruleError != nil {
+ return ruleError
+ }
+ if s.seriesInfo.seriesEvent == nil {
+ return errors.New("No seriesEvents")
+ }
+ seriesIDLen := len(s.seriesInfo.seriesEvent.FieldNamesCompositeSeriesId)
+ var str string
+ var arr []string
+ for i := 0; i < seriesIDLen; i++ {
Review comment:
uh... I am confused. There are two arrays([]*FieldRef and []*Field), can I get results in one iteration? the Field of Write has only Value without Key.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@skywalking.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [skywalking-banyandb] hanahmily commented on a change in pull request #30: feat: Implement Write in the liaison
Posted by GitBox <gi...@apache.org>.
hanahmily commented on a change in pull request #30:
URL: https://github.com/apache/skywalking-banyandb/pull/30#discussion_r685744615
##########
File path: banyand/liaison/grpc/grpc.go
##########
@@ -49,26 +49,35 @@ import (
)
var (
- ErrSeriesEvents = errors.New("no seriesEvent")
- ErrShardEvents = errors.New("no shardEvent")
- ErrInvalidSeriesID = errors.New("invalid seriesID")
+ ErrSeriesEvents = errors.New("no seriesEvent")
+ ErrShardEvents = errors.New("no shardEvent")
+ ErrInvalidSeriesID = errors.New("invalid seriesID")
+ ErrServerCert = errors.New("invalid server cert file")
+ ErrServerKey = errors.New("invalid server key file")
+ ErrServerHostOverride = errors.New("invalid serverHostOverride")
+ ErrNoAddr = errors.New("no address")
)
type Server struct {
- addr string
- log *logger.Logger
- ser *grpclib.Server
- pipeline queue.Queue
- repo discovery.ServiceRepo
- shardInfo *shardInfo
- seriesInfo *seriesInfo
+ addr string
+ maxRecvMsgSize int
+ TLSVal bool
+ ServerHostOverride string
+ CertFile string
Review comment:
If we don't export them, switch them to lower case.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@skywalking.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [skywalking-banyandb] wu-sheng commented on pull request #30: feat: Implement Write in the liaison
Posted by GitBox <gi...@apache.org>.
wu-sheng commented on pull request #30:
URL: https://github.com/apache/skywalking-banyandb/pull/30#issuecomment-894595278
If copy happens, let's keep the original license header, and change license.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@skywalking.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [skywalking-banyandb] Fine0830 commented on pull request #30: feat: Implement Write in the liaison
Posted by GitBox <gi...@apache.org>.
Fine0830 commented on pull request #30:
URL: https://github.com/apache/skywalking-banyandb/pull/30#issuecomment-894917984
Done, please recheck.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@skywalking.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [skywalking-banyandb] lujiajing1126 commented on a change in pull request #30: feat: Implement Write in the liaison
Posted by GitBox <gi...@apache.org>.
lujiajing1126 commented on a change in pull request #30:
URL: https://github.com/apache/skywalking-banyandb/pull/30#discussion_r684878006
##########
File path: banyand/liaison/grpc/grpc.go
##########
@@ -128,3 +158,90 @@ func (s *Server) GracefulStop() {
s.log.Info().Msg("stopping")
s.ser.GracefulStop()
}
+
+func assemblyWriteData(shardID uint, writeEntity *v1.WriteRequest, seriesID uint64) data.TraceWriteDate {
+ return data.TraceWriteDate{ShardID: shardID, SeriesID: seriesID, WriteRequest: writeEntity}
+}
+
+func (s *Server) Write(TraceWriteServer v1.TraceService_WriteServer) error {
+ for {
+ writeEntity, err := TraceWriteServer.Recv()
+ if err == io.EOF {
+ return nil
+ }
+ if err != nil {
+ return err
+ }
+
+ ana := logical.DefaultAnalyzer()
+ metadata := common.Metadata{
+ KindVersion: apischema.SeriesKindVersion,
+ Spec: writeEntity.GetMetadata(),
+ }
+ schema, ruleError := ana.BuildTraceSchema(context.TODO(), metadata)
+ if ruleError != nil {
+ return ruleError
+ }
+ s.seriesInfo.RWMutex.RLock()
Review comment:
You have to release the read lock if it steps into `s.seriesInfo.seriesEvent == nil` condition
##########
File path: banyand/liaison/grpc/grpc.go
##########
@@ -128,3 +158,90 @@ func (s *Server) GracefulStop() {
s.log.Info().Msg("stopping")
s.ser.GracefulStop()
}
+
+func assemblyWriteData(shardID uint, writeEntity *v1.WriteRequest, seriesID uint64) data.TraceWriteDate {
+ return data.TraceWriteDate{ShardID: shardID, SeriesID: seriesID, WriteRequest: writeEntity}
+}
+
+func (s *Server) Write(TraceWriteServer v1.TraceService_WriteServer) error {
+ for {
+ writeEntity, err := TraceWriteServer.Recv()
+ if err == io.EOF {
+ return nil
+ }
+ if err != nil {
+ return err
+ }
+
+ ana := logical.DefaultAnalyzer()
+ metadata := common.Metadata{
+ KindVersion: apischema.SeriesKindVersion,
+ Spec: writeEntity.GetMetadata(),
+ }
+ schema, ruleError := ana.BuildTraceSchema(context.TODO(), metadata)
+ if ruleError != nil {
+ return ruleError
+ }
+ s.seriesInfo.RWMutex.RLock()
+ if s.seriesInfo.seriesEvent == nil {
+ return ErrSeriesEvents
+ }
+ var str string
+ var arr []string
+ fieldRefs, errField := schema.CreateRef(s.seriesInfo.seriesEvent.FieldNamesCompositeSeriesId...)
+ s.seriesInfo.RWMutex.RUnlock()
+ if errField != nil {
+ return errField
+ }
+ for _, ref := range fieldRefs {
+ field := writeEntity.GetEntity().GetFields()[ref.Spec.Idx]
+ switch v := field.GetValueType().(type) {
+ case *v1.Field_StrArray:
+ for j := 0; j < len(v.StrArray.Value); j++ {
+ arr = append(arr, v.StrArray.Value[j])
+ }
+ case *v1.Field_IntArray:
+ for t := 0; t < len(v.IntArray.Value); t++ {
+ arr = append(arr, fmt.Sprint(v.IntArray.Value[t]))
+ }
+ case *v1.Field_Int:
+ arr = append(arr, fmt.Sprint(v.Int.Value))
+ case *v1.Field_Str:
+ arr = append(arr, fmt.Sprint(v.Str.Value))
+ }
+ }
+ str = strings.Join(arr, "")
+ if str == "" {
+ return ErrInvalidSeriesID
+ }
+ seriesID := []byte(str)
+ s.shardInfo.RWMutex.RLock()
Review comment:
Same here
##########
File path: banyand/liaison/grpc/grpc.go
##########
@@ -102,7 +132,9 @@ func (s *Server) Name() string {
}
func (s *Server) FlagSet() *run.FlagSet {
+ size := 1024 * 1024 * 8
fs := run.NewFlagSet("grpc")
+ fs.Int("maxRecMsgSize", size, "the size of max receiving message")
Review comment:
where do we use this variable?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@skywalking.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [skywalking-banyandb] hanahmily commented on a change in pull request #30: feat: Implement Write in the liaison
Posted by GitBox <gi...@apache.org>.
hanahmily commented on a change in pull request #30:
URL: https://github.com/apache/skywalking-banyandb/pull/30#discussion_r686606169
##########
File path: banyand/liaison/grpc/grpc.go
##########
@@ -19,64 +19,146 @@ package grpc
import (
"context"
+ "fmt"
+ "io"
+ "log"
"net"
+ "path/filepath"
+ "runtime"
+ "strings"
+ "sync"
+ "time"
- "google.golang.org/grpc"
+ "github.com/pkg/errors"
+ grpclib "google.golang.org/grpc"
+ "google.golang.org/grpc/credentials"
+ "github.com/apache/skywalking-banyandb/api/common"
+ "github.com/apache/skywalking-banyandb/api/data"
"github.com/apache/skywalking-banyandb/api/event"
v1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/v1"
+ apischema "github.com/apache/skywalking-banyandb/api/schema"
"github.com/apache/skywalking-banyandb/banyand/discovery"
"github.com/apache/skywalking-banyandb/banyand/queue"
"github.com/apache/skywalking-banyandb/pkg/bus"
+ "github.com/apache/skywalking-banyandb/pkg/convert"
"github.com/apache/skywalking-banyandb/pkg/logger"
+ "github.com/apache/skywalking-banyandb/pkg/partition"
+ "github.com/apache/skywalking-banyandb/pkg/query/logical"
"github.com/apache/skywalking-banyandb/pkg/run"
)
+var (
+ ErrSeriesEvents = errors.New("no seriesEvent")
+ ErrShardEvents = errors.New("no shardEvent")
+ ErrInvalidSeriesID = errors.New("invalid seriesID")
+ ErrServerCert = errors.New("invalid server cert file")
+ ErrServerKey = errors.New("invalid server key file")
+ ErrServerHostOverride = errors.New("invalid serverHostOverride")
+ ErrNoAddr = errors.New("no address")
+)
+
type Server struct {
- addr string
- log *logger.Logger
- ser *grpc.Server
- pipeline queue.Queue
- repo discovery.ServiceRepo
- shardInfo *shardInfo
- seriesInfo *seriesInfo
+ addr string
+ maxRecvMsgSize int
+ TLSVal bool
+ ServerHostOverride string
+ CertFile string
+ keyFile string
+ log *logger.Logger
+ ser *grpclib.Server
+ pipeline queue.Queue
+ repo discovery.ServiceRepo
+ shardInfo *shardInfo
+ seriesInfo *seriesInfo
+ v1.UnimplementedTraceServiceServer
}
type shardInfo struct {
- log *logger.Logger
+ log *logger.Logger
+ shardEvent *shardEvent
+ sync.RWMutex
}
func (s *shardInfo) Rev(message bus.Message) (resp bus.Message) {
- shardEvent, ok := message.Data().(*v1.ShardEvent)
+ event, ok := message.Data().(*v1.ShardEvent)
if !ok {
s.log.Warn().Msg("invalid event data type")
return
}
+ s.shardEvent.setShardEvents(event)
s.log.Info().
- Str("action", v1.Action_name[int32(shardEvent.Action)]).
- Uint64("shardID", shardEvent.Shard.Id).
+ Str("action", v1.Action_name[int32(event.Action)]).
+ Uint64("shardID", event.Shard.Id).
Msg("received a shard event")
return
}
+func (s *shardInfo) getShardEvent(idx string) *v1.ShardEvent {
+ s.RWMutex.RLock()
Review comment:
Move `getShardEvent` into `shardEvent` and remove `shardInfo.RWMutex`, that could help us use a single lock to protect shardEventsMap.
##########
File path: banyand/liaison/grpc/grpc.go
##########
@@ -49,26 +49,35 @@ import (
)
var (
- ErrSeriesEvents = errors.New("no seriesEvent")
- ErrShardEvents = errors.New("no shardEvent")
- ErrInvalidSeriesID = errors.New("invalid seriesID")
+ ErrSeriesEvents = errors.New("no seriesEvent")
+ ErrShardEvents = errors.New("no shardEvent")
+ ErrInvalidSeriesID = errors.New("invalid seriesID")
+ ErrServerCert = errors.New("invalid server cert file")
+ ErrServerKey = errors.New("invalid server key file")
+ ErrServerHostOverride = errors.New("invalid serverHostOverride")
+ ErrNoAddr = errors.New("no address")
)
type Server struct {
- addr string
- log *logger.Logger
- ser *grpclib.Server
- pipeline queue.Queue
- repo discovery.ServiceRepo
- shardInfo *shardInfo
- seriesInfo *seriesInfo
+ addr string
+ maxRecvMsgSize int
+ TLSVal bool
+ ServerHostOverride string
+ CertFile string
Review comment:
You could follow the below piece to set those flags instead of exporting them:
```go
tcp.FlagSet().Parse([]string{"--tls=" + xxx})
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@skywalking.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [skywalking-banyandb] Fine0830 commented on a change in pull request #30: feat: Implement Write in the liaison
Posted by GitBox <gi...@apache.org>.
Fine0830 commented on a change in pull request #30:
URL: https://github.com/apache/skywalking-banyandb/pull/30#discussion_r685706364
##########
File path: banyand/liaison/grpc/testdata/server_cert.pem
##########
@@ -0,0 +1,32 @@
+-----BEGIN CERTIFICATE-----
Review comment:
The `server_cert.pem` and `server_key.pem` are needed in server, the `server_cert.pem` is needed in client. The license of cert and key files are Apache-2.0 License, the same as `skywalking-banyandb`, I need to add they?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@skywalking.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [skywalking-banyandb] Fine0830 commented on a change in pull request #30: feat: Implement Write in the liaison
Posted by GitBox <gi...@apache.org>.
Fine0830 commented on a change in pull request #30:
URL: https://github.com/apache/skywalking-banyandb/pull/30#discussion_r685157606
##########
File path: banyand/liaison/grpc/grpc.go
##########
@@ -102,7 +132,9 @@ func (s *Server) Name() string {
}
func (s *Server) FlagSet() *run.FlagSet {
+ size := 1024 * 1024 * 8
fs := run.NewFlagSet("grpc")
+ fs.Int("maxRecMsgSize", size, "the size of max receiving message")
Review comment:
Done. It's added in server options.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@skywalking.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [skywalking-banyandb] lujiajing1126 commented on a change in pull request #30: feat: Implement Write in the liaison
Posted by GitBox <gi...@apache.org>.
lujiajing1126 commented on a change in pull request #30:
URL: https://github.com/apache/skywalking-banyandb/pull/30#discussion_r684577263
##########
File path: banyand/liaison/grpc/grpc.go
##########
@@ -128,3 +166,87 @@ func (s *Server) GracefulStop() {
s.log.Info().Msg("stopping")
s.ser.GracefulStop()
}
+
+func assemblyWriteData(shardID uint, writeEntity *v1.WriteRequest, seriesID uint64) data.TraceWriteDate {
+ return data.TraceWriteDate{ShardID: shardID, SeriesID: seriesID, WriteRequest: writeEntity}
+}
+
+func (s *Server) Write(TraceWriteServer v1.TraceService_WriteServer) error {
+ for {
+ writeEntity, err := TraceWriteServer.Recv()
+ if err == io.EOF {
+ return nil
+ }
+ if err != nil {
+ return err
+ }
+
+ //log.Println("writeEntity:", writeEntity)
+ ana := logical.DefaultAnalyzer()
+ metadata := common.Metadata{
+ KindVersion: apischema.SeriesKindVersion,
+ Spec: writeEntity.GetMetadata(),
+ }
+ schema, ruleError := ana.BuildTraceSchema(context.TODO(), metadata)
+ if ruleError != nil {
+ return ruleError
+ }
+ if s.seriesInfo.seriesEvent == nil {
+ return errors.New("No seriesEvents")
+ }
+ seriesIDLen := len(s.seriesInfo.seriesEvent.FieldNamesCompositeSeriesId)
+ var str string
+ var arr []string
+ for i := 0; i < seriesIDLen; i++ {
Review comment:
I suggest optimizing this loop by,
1) first convert `FieldNamesCompositeSeriesId` to []*FieldRef (i.e. `FieldSubscript` in your case, maybe you can make use of `Schema.CreateRef(names ...string)` to do this),
2) then iterate over fields (`writeEntity.GetEntity().GetFields()`) to compose the seriesID,
If we do so, only one iteration is necessary.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@skywalking.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [skywalking-banyandb] lujiajing1126 commented on a change in pull request #30: feat: Implement Write in the liaison
Posted by GitBox <gi...@apache.org>.
lujiajing1126 commented on a change in pull request #30:
URL: https://github.com/apache/skywalking-banyandb/pull/30#discussion_r684573434
##########
File path: banyand/liaison/data/x509/ca_cert.pem
##########
@@ -0,0 +1,34 @@
+-----BEGIN CERTIFICATE-----
Review comment:
Do we need to include certs in the code base?
I suppose it would be better if they can be generated from `Makefile`?
##########
File path: banyand/liaison/grpc/grpc.go
##########
@@ -19,31 +19,54 @@ package grpc
import (
"context"
+ "flag"
+ "fmt"
+ "io"
+ "log"
"net"
+ "strings"
+ "time"
- "google.golang.org/grpc"
+ "github.com/pkg/errors"
+ grpclib "google.golang.org/grpc"
+ "google.golang.org/grpc/credentials"
+ "github.com/apache/skywalking-banyandb/api/common"
+ "github.com/apache/skywalking-banyandb/api/data"
"github.com/apache/skywalking-banyandb/api/event"
v1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/v1"
+ apischema "github.com/apache/skywalking-banyandb/api/schema"
"github.com/apache/skywalking-banyandb/banyand/discovery"
+ serverData "github.com/apache/skywalking-banyandb/banyand/liaison/data"
"github.com/apache/skywalking-banyandb/banyand/queue"
"github.com/apache/skywalking-banyandb/pkg/bus"
+ "github.com/apache/skywalking-banyandb/pkg/convert"
"github.com/apache/skywalking-banyandb/pkg/logger"
+ "github.com/apache/skywalking-banyandb/pkg/partition"
+ "github.com/apache/skywalking-banyandb/pkg/query/logical"
"github.com/apache/skywalking-banyandb/pkg/run"
)
+var (
+ tls = flag.Bool("tls", false, "Connection uses TLS if true, else plain TCP")
Review comment:
As I understand, the flags should be defined in the `func (s *Server) FlagSet() *run.FlagSet`? @hanahmily
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@skywalking.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org