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 2021/08/13 02:08:45 UTC
[skywalking-banyandb] branch main updated: feat: Implement Query in
the liaison (#33)
This is an automated email from the ASF dual-hosted git repository.
hanahmily pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/skywalking-banyandb.git
The following commit(s) were added to refs/heads/main by this push:
new d5061ee feat: Implement Query in the liaison (#33)
d5061ee is described below
commit d5061eebeaccf685e0070adfc11b4371b5f17721
Author: Fine0830 <fi...@outlook.com>
AuthorDate: Fri Aug 13 10:08:40 2021 +0800
feat: Implement Query in the liaison (#33)
* feat: update query topic
* feat: implement query
* feat: update query msg
* fix: update test
* fix: update type
* fix: lint
* fix: address pr
* fix: update grpc test
* fix: update grpc test
* fix: update test data
---
api/data/trace.go | 6 ++
api/event/query.go | 33 --------
banyand/internal/cmd/standalone.go | 2 +-
banyand/liaison/grpc/grpc.go | 22 ++++-
banyand/liaison/grpc/grpc_test.go | 161 ++++++++++++++++++++++++-------------
banyand/query/processor.go | 6 +-
banyand/query/processor_test.go | 12 +--
banyand/query/query.go | 5 +-
8 files changed, 143 insertions(+), 104 deletions(-)
diff --git a/api/data/trace.go b/api/data/trace.go
index 6d59308..0f8b84d 100644
--- a/api/data/trace.go
+++ b/api/data/trace.go
@@ -31,6 +31,12 @@ var WriteEventKindVersion = common.KindVersion{
}
var TopicWriteEvent = bus.UniTopic(WriteEventKindVersion.String())
+var QueryEventKindVersion = common.KindVersion{
+ Version: "v1",
+ Kind: "trace-query",
+}
+var TopicQueryEvent = bus.BiTopic(QueryEventKindVersion.String())
+
type Trace struct {
common.KindVersion
Entities []Entity
diff --git a/api/event/query.go b/api/event/query.go
deleted file mode 100644
index be842a9..0000000
--- a/api/event/query.go
+++ /dev/null
@@ -1,33 +0,0 @@
-// 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 event
-
-import (
- "github.com/apache/skywalking-banyandb/api/common"
- "github.com/apache/skywalking-banyandb/pkg/bus"
-)
-
-var (
- QueryEventKindVersion = common.KindVersion{
- Version: "v1",
- Kind: "event-query",
- }
- // TopicQueryEvent is a bidirectional topic for request/response communication
- // between Liaison and Query module
- TopicQueryEvent = bus.BiTopic(QueryEventKindVersion.String())
-)
diff --git a/banyand/internal/cmd/standalone.go b/banyand/internal/cmd/standalone.go
index 1aa0b4f..5ee6c22 100644
--- a/banyand/internal/cmd/standalone.go
+++ b/banyand/internal/cmd/standalone.go
@@ -65,7 +65,7 @@ func newStandaloneCmd() *cobra.Command {
if err != nil {
l.Fatal().Err(err).Msg("failed to initiate trace series")
}
- q, err := query.NewExecutor(ctx, repo, idx, traceSeries, traceSeries)
+ q, err := query.NewExecutor(ctx, repo, idx, traceSeries, traceSeries, pipeline)
if err != nil {
l.Fatal().Err(err).Msg("failed to initiate query executor")
}
diff --git a/banyand/liaison/grpc/grpc.go b/banyand/liaison/grpc/grpc.go
index 39451c5..2ea2623 100644
--- a/banyand/liaison/grpc/grpc.go
+++ b/banyand/liaison/grpc/grpc.go
@@ -21,7 +21,6 @@ import (
"context"
"fmt"
"io"
- "log"
"net"
"path/filepath"
"runtime"
@@ -55,6 +54,7 @@ var (
ErrServerCert = errors.New("invalid server cert file")
ErrServerKey = errors.New("invalid server key file")
ErrNoAddr = errors.New("no address")
+ ErrQueryMsg = errors.New("invalid query message")
)
type Server struct {
@@ -334,9 +334,25 @@ func (s *Server) Write(TraceWriteServer v1.TraceService_WriteServer) error {
}
func (s *Server) Query(ctx context.Context, entityCriteria *v1.QueryRequest) (*v1.QueryResponse, error) {
- log.Println("entityCriteria:", entityCriteria)
+ message := bus.NewMessage(bus.MessageID(time.Now().UnixNano()), entityCriteria)
+ feat, errQuery := s.pipeline.Publish(data.TopicQueryEvent, message)
+ if errQuery != nil {
+ return nil, errQuery
+ }
+ msg, errFeat := feat.Get()
+ if errFeat != nil {
+ return nil, errFeat
+ }
+ queryMsg, ok := msg.Data().([]data.Entity)
+ if !ok {
+ return nil, ErrQueryMsg
+ }
+ var arr []*v1.Entity
+ for i := 0; i < len(queryMsg); i++ {
+ arr = append(arr, queryMsg[i].Entity)
+ }
- return &v1.QueryResponse{}, nil
+ return &v1.QueryResponse{Entities: arr}, nil
}
func assemblyWriteData(shardID uint, writeEntity *v1.WriteRequest, seriesID uint64) data.TraceWriteDate {
diff --git a/banyand/liaison/grpc/grpc_test.go b/banyand/liaison/grpc/grpc_test.go
index 5aea70f..7b13bdd 100644
--- a/banyand/liaison/grpc/grpc_test.go
+++ b/banyand/liaison/grpc/grpc_test.go
@@ -78,7 +78,7 @@ func setup(tester *require.Assertions) (*grpc.Server, *grpc.Server, func()) {
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)
+ executor, err := query.NewExecutor(context.TODO(), repo, indexSvc, traceSvc, traceSvc, pipeline)
tester.NoError(err)
// Init `liaison` module
tcp := grpc.NewServer(context.TODO(), pipeline, repo)
@@ -125,6 +125,14 @@ func setup(tester *require.Assertions) (*grpc.Server, *grpc.Server, func()) {
}
}
+type caseData struct {
+ name string
+ queryGenerator func(baseTs time.Time) *v1.QueryRequest
+ writeGenerator func() *v1.WriteRequest
+ args testData
+ wantLen int
+}
+
func TestTraceService(t *testing.T) {
tester := require.New(t)
tcp, tcpTLS, gracefulStop := setup(tester)
@@ -132,77 +140,122 @@ func TestTraceService(t *testing.T) {
_, currentFile, _, _ := runtime.Caller(0)
basePath := filepath.Dir(currentFile)
certFile := filepath.Join(basePath, "data/server_cert.pem")
- testCases := []struct {
- name string
- args testData
- wantErr bool
- }{
+ testCases := []caseData{
{
name: "isTLS",
+ queryGenerator: func(baseTs time.Time) *v1.QueryRequest {
+ return pb.NewQueryRequestBuilder().
+ Limit(10).
+ Offset(0).
+ Metadata("default", "sw").
+ Fields("trace_id", "=", "trace_id-xxfff.111").
+ TimeRange(baseTs.Add(-1*time.Minute), baseTs.Add(1*time.Minute)).
+ Projection("trace_id").
+ Build()
+ },
+ writeGenerator: func() *v1.WriteRequest {
+ entityValue := pb.NewEntityValueBuilder().
+ EntityID("entityId123").
+ DataBinary([]byte{12}).
+ Fields("trace_id-xxfff.111",
+ 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()
+ return criteria
+ },
args: testData{
TLS: true,
certFile: certFile,
serverHostOverride: "localhost",
addr: "localhost:17913",
},
+ wantLen: 1,
},
{
name: "noTLS",
+ queryGenerator: func(baseTs time.Time) *v1.QueryRequest {
+ return pb.NewQueryRequestBuilder().
+ Limit(10).
+ Offset(0).
+ Metadata("default", "sw").
+ Fields("trace_id", "=", "trace_id-xxfff.111323").
+ TimeRange(baseTs.Add(-1*time.Minute), baseTs.Add(1*time.Minute)).
+ Projection("trace_id").
+ Build()
+ },
+ writeGenerator: func() *v1.WriteRequest {
+ entityValue := pb.NewEntityValueBuilder().
+ EntityID("entityId123").
+ 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()
+ return criteria
+ },
args: testData{
TLS: false,
addr: "localhost:17912",
},
+ wantLen: 1,
},
}
for _, tc := range testCases {
- if tc.args.TLS {
- errValidate := tcpTLS.Validate()
- assert.NoError(t, errValidate)
- var opts []grpclib.DialOption
- creds, err := credentials.NewClientTLSFromFile(tc.args.certFile, tc.args.serverHostOverride)
- assert.NoError(t, err)
- opts = append(opts, grpclib.WithTransportCredentials(creds))
- linkService(t, tc.args.addr, opts)
- } else {
- errValidate := tcp.Validate()
- assert.NoError(t, errValidate)
- var opts []grpclib.DialOption
- opts = append(opts, grpclib.WithInsecure())
- linkService(t, tc.args.addr, opts)
- }
+ t.Run(tc.name, func(t *testing.T) {
+ if tc.args.TLS {
+ errValidate := tcpTLS.Validate()
+ assert.NoError(t, errValidate)
+ var opts []grpclib.DialOption
+ creds, err := credentials.NewClientTLSFromFile(tc.args.certFile, tc.args.serverHostOverride)
+ assert.NoError(t, err)
+ opts = append(opts, grpclib.WithTransportCredentials(creds))
+ dialService(t, tc, opts)
+ } else {
+ errValidate := tcp.Validate()
+ assert.NoError(t, errValidate)
+ var opts []grpclib.DialOption
+ opts = append(opts, grpclib.WithInsecure())
+ dialService(t, tc, opts)
+ }
+ })
}
}
-func linkService(t *testing.T, addr string, opts []grpclib.DialOption) {
- conn, err := grpclib.Dial(addr, opts...)
+func dialService(t *testing.T, tc caseData, opts []grpclib.DialOption) {
+ conn, err := grpclib.Dial(tc.args.addr, opts...)
assert.NoError(t, err)
defer conn.Close()
- traceWrite(t, conn)
- traceQuery(t, conn)
+ traceWrite(t, tc, conn)
+ traceQuery(t, tc, conn)
}
-func traceWrite(t *testing.T, conn *grpclib.ClientConn) {
+func traceWrite(t *testing.T, tc caseData, conn *grpclib.ClientConn) {
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 {
t.Errorf("%v.Write(_) = _, %v", client, errorWrite)
@@ -220,7 +273,7 @@ func traceWrite(t *testing.T, conn *grpclib.ClientConn) {
assert.NotNil(t, writeResponse)
}
}()
- if errSend := stream.Send(criteria); errSend != nil {
+ if errSend := stream.Send(tc.writeGenerator()); errSend != nil {
t.Errorf("Failed to send a note: %v", errSend)
}
if errorSend := stream.CloseSend(); errorSend != nil {
@@ -229,22 +282,14 @@ func traceWrite(t *testing.T, conn *grpclib.ClientConn) {
<-waitc
}
-func traceQuery(t *testing.T, conn *grpclib.ClientConn) {
+func traceQuery(t *testing.T, tc caseData, conn *grpclib.ClientConn) {
client := v1.NewTraceServiceClient(conn)
ctx := context.Background()
- sT, eT := time.Now().Add(-3*time.Hour), time.Now()
- criteria := pb.NewQueryRequestBuilder().
- Limit(5).
- Offset(10).
- OrderBy("service_instance_id", v1.QueryOrder_SORT_DESC).
- Metadata("default", "trace").
- Projection("http.method", "service_id", "service_instance_id").
- Fields("service_id", "=", "my_app", "http.method", "=", "GET").
- TimeRange(sT, eT).
- Build()
- stream, errRev := client.Query(ctx, criteria)
+ baseTs := time.Now()
+ stream, errRev := client.Query(ctx, tc.queryGenerator(baseTs))
if errRev != nil {
t.Errorf("Retrieve client failed: %v", errRev)
}
assert.NotNil(t, stream)
+ assert.Len(t, stream.Entities, tc.wantLen)
}
diff --git a/banyand/query/processor.go b/banyand/query/processor.go
index 2bbb121..807fcc7 100644
--- a/banyand/query/processor.go
+++ b/banyand/query/processor.go
@@ -22,11 +22,12 @@ import (
"time"
"github.com/apache/skywalking-banyandb/api/common"
- "github.com/apache/skywalking-banyandb/api/event"
+ "github.com/apache/skywalking-banyandb/api/data"
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/index"
+ "github.com/apache/skywalking-banyandb/banyand/queue"
"github.com/apache/skywalking-banyandb/banyand/series"
"github.com/apache/skywalking-banyandb/pkg/bus"
"github.com/apache/skywalking-banyandb/pkg/logger"
@@ -51,6 +52,7 @@ type queryProcessor struct {
schemaRepo series.SchemaRepo
log *logger.Logger
serviceRepo discovery.ServiceRepo
+ pipeline queue.Queue
}
func (q *queryProcessor) Rev(message bus.Message) (resp bus.Message) {
@@ -96,5 +98,5 @@ func (q *queryProcessor) Name() string {
func (q *queryProcessor) PreRun() error {
q.log = logger.GetLogger(moduleName)
- return q.serviceRepo.Subscribe(event.TopicQueryEvent, q)
+ return q.pipeline.Subscribe(data.TopicQueryEvent, q)
}
diff --git a/banyand/query/processor_test.go b/banyand/query/processor_test.go
index f3da0a2..a1abe7b 100644
--- a/banyand/query/processor_test.go
+++ b/banyand/query/processor_test.go
@@ -28,7 +28,7 @@ import (
"github.com/stretchr/testify/require"
"github.com/apache/skywalking-banyandb/api/common"
- "github.com/apache/skywalking-banyandb/api/event"
+ "github.com/apache/skywalking-banyandb/api/data"
v1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/v1"
"github.com/apache/skywalking-banyandb/banyand/discovery"
"github.com/apache/skywalking-banyandb/banyand/index"
@@ -52,7 +52,7 @@ type entityValue struct {
items []interface{}
}
-func setupServices(t *testing.T, tester *require.Assertions) (discovery.ServiceRepo, series.Service, func()) {
+func setupServices(t *testing.T, tester *require.Assertions) (series.Service, queue.Queue, func()) {
// Bootstrap logger system
tester.NoError(logger.Init(logger.Logging{
Env: "dev",
@@ -84,7 +84,7 @@ func setupServices(t *testing.T, tester *require.Assertions) (discovery.ServiceR
tester.NoError(err)
// Init `Query` module
- executor, err := NewExecutor(context.TODO(), repo, indexSvc, traceSvc, traceSvc)
+ executor, err := NewExecutor(context.TODO(), repo, indexSvc, traceSvc, traceSvc, pipeline)
tester.NoError(err)
// Init `Liaison` module
@@ -119,7 +119,7 @@ func setupServices(t *testing.T, tester *require.Assertions) (discovery.ServiceR
defer cancelFunc()
tester.True(indexSvc.Ready(ctx, index.MetaExists("default", "sw")))
- return repo, traceSvc, func() {
+ return traceSvc, pipeline, func() {
db.GracefulStop()
_ = os.RemoveAll(rootPath)
}
@@ -287,7 +287,7 @@ func TestQueryProcessor(t *testing.T) {
tester := require.New(t)
// setup services
- repo, traceSvc, gracefulStop := setupServices(t, tester)
+ traceSvc, pipeline, gracefulStop := setupServices(t, tester)
defer gracefulStop()
baseTs := time.Now()
@@ -418,7 +418,7 @@ func TestQueryProcessor(t *testing.T) {
singleTester := require.New(t)
now := time.Now()
m := bus.NewMessage(bus.MessageID(now.UnixNano()), tt.queryGenerator(baseTs))
- f, err := repo.Publish(event.TopicQueryEvent, m)
+ f, err := pipeline.Publish(data.TopicQueryEvent, m)
singleTester.NoError(err)
singleTester.NotNil(f)
msg, err := f.Get()
diff --git a/banyand/query/query.go b/banyand/query/query.go
index 3c3f9a8..cbc990f 100644
--- a/banyand/query/query.go
+++ b/banyand/query/query.go
@@ -22,6 +22,7 @@ import (
"github.com/apache/skywalking-banyandb/banyand/discovery"
"github.com/apache/skywalking-banyandb/banyand/index"
+ "github.com/apache/skywalking-banyandb/banyand/queue"
"github.com/apache/skywalking-banyandb/banyand/series"
"github.com/apache/skywalking-banyandb/pkg/logger"
"github.com/apache/skywalking-banyandb/pkg/run"
@@ -31,12 +32,14 @@ type Executor interface {
run.PreRunner
}
-func NewExecutor(_ context.Context, serviceRepo discovery.ServiceRepo, indexRepo index.Repo, uniModel series.UniModel, schemaRepo series.SchemaRepo) (Executor, error) {
+func NewExecutor(_ context.Context, serviceRepo discovery.ServiceRepo, indexRepo index.Repo, uniModel series.UniModel,
+ schemaRepo series.SchemaRepo, pipeline queue.Queue) (Executor, error) {
return &queryProcessor{
Repo: indexRepo,
UniModel: uniModel,
schemaRepo: schemaRepo,
serviceRepo: serviceRepo,
logger: logger.GetLogger("query"),
+ pipeline: pipeline,
}, nil
}