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 2025/09/05 06:58:23 UTC
(skywalking-banyandb) 01/02: feat: add query access logging to gRPC services
This is an automated email from the ASF dual-hosted git repository.
hanahmily pushed a commit to branch feat/query-accesslog
in repository https://gitbox.apache.org/repos/asf/skywalking-banyandb.git
commit 23e052970d44ce81ca29749fcae309c9628dcace
Author: Gao Hongtao <ha...@gmail.com>
AuthorDate: Fri Sep 5 14:20:23 2025 +0800
feat: add query access logging to gRPC services
- Introduced query access logging functionality across multiple gRPC services including measure, property, stream, and trace.
- Implemented methods to activate and write to query access logs, enhancing observability and debugging capabilities.
- Updated server configuration to enable query access logging and manage associated log recorders.
---
banyand/liaison/grpc/measure.go | 19 +++++++++++++
banyand/liaison/grpc/property.go | 61 ++++++++++++++++++++++++++++++++++++----
banyand/liaison/grpc/server.go | 24 +++++++++++++++-
banyand/liaison/grpc/stream.go | 14 +++++++++
banyand/liaison/grpc/trace.go | 16 ++++++++++-
5 files changed, 126 insertions(+), 8 deletions(-)
diff --git a/banyand/liaison/grpc/measure.go b/banyand/liaison/grpc/measure.go
index 373a676c..b3d6912b 100644
--- a/banyand/liaison/grpc/measure.go
+++ b/banyand/liaison/grpc/measure.go
@@ -44,6 +44,7 @@ import (
type measureService struct {
measurev1.UnimplementedMeasureServiceServer
ingestionAccessLog accesslog.Log
+ queryAccessLog accesslog.Log
pipeline queue.Client
broadcaster queue.Client
*discoveryService
@@ -65,6 +66,14 @@ func (ms *measureService) activeIngestionAccessLog(root string, sampled bool) (e
return nil
}
+func (ms *measureService) activeQueryAccessLog(root string, sampled bool) (err error) {
+ if ms.queryAccessLog, err = accesslog.
+ NewFileLog(root, "measure-query-%s", 10*time.Minute, ms.log, sampled); err != nil {
+ return err
+ }
+ return nil
+}
+
func (ms *measureService) Write(measure measurev1.MeasureService_WriteServer) error {
ctx := measure.Context()
publisher := ms.pipeline.NewBatchPublisher(ms.writeTimeout)
@@ -277,6 +286,11 @@ func (ms *measureService) Query(ctx context.Context, req *measurev1.QueryRequest
ms.metrics.totalLatency.Inc(time.Since(start).Seconds(), g, "measure", "query")
}
}()
+ if ms.queryAccessLog != nil {
+ if errAccessLog := ms.queryAccessLog.Write(req); errAccessLog != nil {
+ ms.l.Error().Err(errAccessLog).Msg("query access log error")
+ }
+ }
if err = timestamp.CheckTimeRange(req.GetTimeRange()); err != nil {
return nil, status.Errorf(codes.InvalidArgument, "%v is invalid :%s", req.GetTimeRange(), err)
}
@@ -317,6 +331,11 @@ func (ms *measureService) Query(ctx context.Context, req *measurev1.QueryRequest
}
func (ms *measureService) TopN(ctx context.Context, topNRequest *measurev1.TopNRequest) (resp *measurev1.TopNResponse, err error) {
+ if ms.queryAccessLog != nil {
+ if errAccessLog := ms.queryAccessLog.Write(topNRequest); errAccessLog != nil {
+ ms.l.Error().Err(errAccessLog).Msg("query access log error")
+ }
+ }
if err = timestamp.CheckTimeRange(topNRequest.GetTimeRange()); err != nil {
return nil, status.Errorf(codes.InvalidArgument, "%v is invalid :%s", topNRequest.GetTimeRange(), err)
}
diff --git a/banyand/liaison/grpc/property.go b/banyand/liaison/grpc/property.go
index cd4921c7..99880b86 100644
--- a/banyand/liaison/grpc/property.go
+++ b/banyand/liaison/grpc/property.go
@@ -39,6 +39,7 @@ import (
"github.com/apache/skywalking-banyandb/banyand/metadata/schema"
propertypkg "github.com/apache/skywalking-banyandb/banyand/property"
"github.com/apache/skywalking-banyandb/banyand/queue"
+ "github.com/apache/skywalking-banyandb/pkg/accesslog"
"github.com/apache/skywalking-banyandb/pkg/bus"
"github.com/apache/skywalking-banyandb/pkg/convert"
"github.com/apache/skywalking-banyandb/pkg/logger"
@@ -52,12 +53,45 @@ const defaultQueryTimeout = 10 * time.Second
type propertyServer struct {
propertyv1.UnimplementedPropertyServiceServer
*discoveryService
- schemaRegistry metadata.Repo
- pipeline queue.Client
- nodeRegistry NodeRegistry
- metrics *metrics
- repairQueue *repairQueue
- repairQueueCount int
+ ingestionAccessLog accesslog.Log
+ queryAccessLog accesslog.Log
+ schemaRegistry metadata.Repo
+ pipeline queue.Client
+ nodeRegistry NodeRegistry
+ metrics *metrics
+ repairQueue *repairQueue
+ repairQueueCount int
+}
+
+func (ps *propertyServer) activeIngestionAccessLog(root string, sampled bool) (err error) {
+ if ps.ingestionAccessLog, err = accesslog.
+ NewFileLog(root, "property-ingest-%s", 10*time.Minute, ps.log, sampled); err != nil {
+ return err
+ }
+ return nil
+}
+
+func (ps *propertyServer) activeQueryAccessLog(root string, sampled bool) (err error) {
+ if ps.queryAccessLog, err = accesslog.
+ NewFileLog(root, "property-query-%s", 10*time.Minute, ps.log, sampled); err != nil {
+ return err
+ }
+ return nil
+}
+
+func (ps *propertyServer) Close() error {
+ var err error
+ if ps.ingestionAccessLog != nil {
+ if closeErr := ps.ingestionAccessLog.Close(); closeErr != nil {
+ err = closeErr
+ }
+ }
+ if ps.queryAccessLog != nil {
+ if closeErr := ps.queryAccessLog.Close(); closeErr != nil {
+ err = closeErr
+ }
+ }
+ return err
}
func (ps *propertyServer) validatePropertyRequest(property *propertyv1.Property) error {
@@ -163,6 +197,11 @@ func (ps *propertyServer) Apply(ctx context.Context, req *propertyv1.ApplyReques
ps.metrics.totalErr.Inc(1, g, "property", "apply")
}
}()
+ if ps.ingestionAccessLog != nil {
+ if errAccessLog := ps.ingestionAccessLog.Write(req); errAccessLog != nil {
+ ps.log.Error().Err(errAccessLog).Msg("ingestion access log error")
+ }
+ }
var group *commonv1.Group
if group, err = ps.validateGroupForProperty(ctx, g); err != nil {
return nil, err
@@ -337,6 +376,11 @@ func (ps *propertyServer) Delete(ctx context.Context, req *propertyv1.DeleteRequ
ps.metrics.totalErr.Inc(1, g, "property", "delete")
}
}()
+ if ps.ingestionAccessLog != nil {
+ if errAccessLog := ps.ingestionAccessLog.Write(req); errAccessLog != nil {
+ ps.log.Error().Err(errAccessLog).Msg("ingestion access log error")
+ }
+ }
qReq := &propertyv1.QueryRequest{
Groups: []string{g},
Name: req.Name,
@@ -377,6 +421,11 @@ func (ps *propertyServer) Query(ctx context.Context, req *propertyv1.QueryReques
ps.metrics.totalErr.Inc(1, "", "property", "query")
}
}()
+ if ps.queryAccessLog != nil {
+ if errAccessLog := ps.queryAccessLog.Write(req); errAccessLog != nil {
+ ps.log.Error().Err(errAccessLog).Msg("query access log error")
+ }
+ }
if len(req.Groups) == 0 {
return nil, schema.BadRequest("groups", "groups should not be empty")
}
diff --git a/banyand/liaison/grpc/server.go b/banyand/liaison/grpc/server.go
index 762ecdb9..2c4d13e1 100644
--- a/banyand/liaison/grpc/server.go
+++ b/banyand/liaison/grpc/server.go
@@ -108,10 +108,12 @@ type server struct {
accessLogRootPath string
certFile string
accessLogRecorders []accessLogRecorder
+ queryAccessLogRecorders []queryAccessLogRecorder
maxRecvMsgSize run.Bytes
port uint32
tls bool
enableIngestionAccessLog bool
+ enableQueryAccessLog bool
accessLogSampled bool
healthAuthEnabled bool
}
@@ -177,7 +179,8 @@ func NewServer(_ context.Context, tir1Client, tir2Client, broadcaster queue.Clie
schemaRepo: schemaRegistry,
authReloader: auth.InitAuthReloader(),
}
- s.accessLogRecorders = []accessLogRecorder{streamSVC, measureSVC, traceSVC}
+ s.accessLogRecorders = []accessLogRecorder{streamSVC, measureSVC, traceSVC, s.propertyServer}
+ s.queryAccessLogRecorders = []queryAccessLogRecorder{streamSVC, measureSVC, traceSVC, s.propertyServer}
return s
}
@@ -214,6 +217,14 @@ func (s *server) PreRun(_ context.Context) error {
}
}
}
+
+ if s.enableQueryAccessLog {
+ for _, qalr := range s.queryAccessLogRecorders {
+ if err := qalr.activeQueryAccessLog(s.accessLogRootPath, s.accessLogSampled); err != nil {
+ return err
+ }
+ }
+ }
metrics := newMetrics(s.omr.With(liaisonGrpcScope))
s.metrics = metrics
s.streamSVC.metrics = metrics
@@ -269,6 +280,7 @@ func (s *server) FlagSet() *run.FlagSet {
fs.StringVar(&s.host, "grpc-host", "", "the host of banyand listens")
fs.Uint32Var(&s.port, "grpc-port", 17912, "the port of banyand listens")
fs.BoolVar(&s.enableIngestionAccessLog, "enable-ingestion-access-log", false, "enable ingestion access log")
+ fs.BoolVar(&s.enableQueryAccessLog, "enable-query-access-log", false, "enable query access log")
fs.StringVar(&s.accessLogRootPath, "access-log-root-path", "", "access log root path")
fs.BoolVar(&s.accessLogSampled, "access-log-sampled", false, "if true, requests may be dropped when the channel is full; if false, requests are never dropped")
fs.DurationVar(&s.streamSVC.writeTimeout, "stream-write-timeout", 15*time.Second, "timeout for writing stream among liaison nodes")
@@ -399,6 +411,11 @@ func (s *server) GracefulStop() {
_ = alr.Close()
}
}
+ if s.enableQueryAccessLog {
+ for _, qalr := range s.queryAccessLogRecorders {
+ _ = qalr.Close()
+ }
+ }
close(stopped)
}()
@@ -417,3 +434,8 @@ type accessLogRecorder interface {
activeIngestionAccessLog(root string, sampled bool) error
Close() error
}
+
+type queryAccessLogRecorder interface {
+ activeQueryAccessLog(root string, sampled bool) error
+ Close() error
+}
diff --git a/banyand/liaison/grpc/stream.go b/banyand/liaison/grpc/stream.go
index b9a6236c..e8e65d13 100644
--- a/banyand/liaison/grpc/stream.go
+++ b/banyand/liaison/grpc/stream.go
@@ -44,6 +44,7 @@ import (
type streamService struct {
streamv1.UnimplementedStreamServiceServer
ingestionAccessLog accesslog.Log
+ queryAccessLog accesslog.Log
pipeline queue.Client
broadcaster queue.Client
*discoveryService
@@ -65,6 +66,14 @@ func (s *streamService) activeIngestionAccessLog(root string, sampled bool) (err
return nil
}
+func (s *streamService) activeQueryAccessLog(root string, sampled bool) (err error) {
+ if s.queryAccessLog, err = accesslog.
+ NewFileLog(root, "stream-query-%s", 10*time.Minute, s.log, sampled); err != nil {
+ return err
+ }
+ return nil
+}
+
func (s *streamService) validateTimestamp(writeEntity *streamv1.WriteRequest) error {
if err := timestamp.CheckPb(writeEntity.GetElement().Timestamp); err != nil {
s.l.Error().Stringer("written", writeEntity).Err(err).Msg("the element time is invalid")
@@ -255,6 +264,11 @@ func (s *streamService) Query(ctx context.Context, req *streamv1.QueryRequest) (
s.metrics.totalLatency.Inc(time.Since(start).Seconds(), g, "stream", "query")
}
}()
+ if s.queryAccessLog != nil {
+ if errAccessLog := s.queryAccessLog.Write(req); errAccessLog != nil {
+ s.l.Error().Err(errAccessLog).Msg("query access log error")
+ }
+ }
timeRange := req.GetTimeRange()
if timeRange == nil {
req.TimeRange = timestamp.DefaultTimeRange
diff --git a/banyand/liaison/grpc/trace.go b/banyand/liaison/grpc/trace.go
index 599d054c..7f265c82 100644
--- a/banyand/liaison/grpc/trace.go
+++ b/banyand/liaison/grpc/trace.go
@@ -44,6 +44,7 @@ import (
type traceService struct {
tracev1.UnimplementedTraceServiceServer
ingestionAccessLog accesslog.Log
+ queryAccessLog accesslog.Log
pipeline queue.Client
broadcaster queue.Client
*discoveryService
@@ -59,7 +60,15 @@ func (s *traceService) setLogger(log *logger.Logger) {
func (s *traceService) activeIngestionAccessLog(root string, sampled bool) (err error) {
if s.ingestionAccessLog, err = accesslog.
- NewFileLog(root, "trace-ingest-%s", 10*time.Minute, s.log, sampled); err != nil {
+ NewFileLog(root, "trace-ingest-%s", 10*time.Minute, s.l, sampled); err != nil {
+ return err
+ }
+ return nil
+}
+
+func (s *traceService) activeQueryAccessLog(root string, sampled bool) (err error) {
+ if s.queryAccessLog, err = accesslog.
+ NewFileLog(root, "trace-query-%s", 10*time.Minute, s.l, sampled); err != nil {
return err
}
return nil
@@ -318,6 +327,11 @@ func (s *traceService) Query(ctx context.Context, req *tracev1.QueryRequest) (re
s.metrics.totalLatency.Inc(time.Since(start).Seconds(), g, "trace", "query")
}
}()
+ if s.queryAccessLog != nil {
+ if errAccessLog := s.queryAccessLog.Write(req); errAccessLog != nil {
+ s.l.Error().Err(errAccessLog).Msg("query access log error")
+ }
+ }
timeRange := req.GetTimeRange()
if timeRange == nil {
req.TimeRange = timestamp.DefaultTimeRange