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/07/19 22:44:49 UTC
(skywalking-banyandb) 01/01: Add trace support with schema, metadata, and validation
This is an automated email from the ASF dual-hosted git repository.
hanahmily pushed a commit to branch trace-module
in repository https://gitbox.apache.org/repos/asf/skywalking-banyandb.git
commit c9c320400babe9b1d666169ed627b1479d7a14e7
Author: Cursor Agent <cu...@cursor.com>
AuthorDate: Fri Jul 18 14:39:02 2025 +0000
Add trace support with schema, metadata, and validation
---
api/validate/validate.go | 35 +++++
banyand/metadata/client.go | 4 +
banyand/metadata/metadata.go | 1 +
banyand/metadata/schema/checker.go | 7 +
banyand/metadata/schema/etcd.go | 1 +
banyand/metadata/schema/kind.go | 15 ++-
banyand/metadata/schema/schema.go | 15 +++
banyand/metadata/schema/trace.go | 83 ++++++++++++
banyand/trace-plan.md | 264 -------------------------------------
banyand/trace/metadata.go | 184 ++++++++++++++++++++++++++
banyand/trace/svc_liaison.go | 190 ++++++++++++++++++++++++++
banyand/trace/svc_standalone.go | 200 ++++++++++++++++++++++++++++
banyand/trace/trace.go | 137 +++++++++++++++++++
13 files changed, 870 insertions(+), 266 deletions(-)
diff --git a/api/validate/validate.go b/api/validate/validate.go
index c8e74bed..d9389a29 100644
--- a/api/validate/validate.go
+++ b/api/validate/validate.go
@@ -167,6 +167,41 @@ func Measure(measure *databasev1.Measure) error {
return tagFamily(measure.TagFamilies)
}
+// Trace validates the provided Trace object.
+// It checks for nil values, empty strings, and unspecified enum values.
+func Trace(trace *databasev1.Trace) error {
+ if trace == nil {
+ return errors.New("trace is nil")
+ }
+ if trace.Metadata == nil {
+ return errors.New("trace metadata is nil")
+ }
+ if trace.Metadata.Name == "" {
+ return errors.New("trace name is empty")
+ }
+ if trace.Metadata.Group == "" {
+ return errors.New("trace group is empty")
+ }
+ if len(trace.Tags) == 0 {
+ return errors.New("trace tags is empty")
+ }
+ if trace.TraceIdTagName == "" {
+ return errors.New("trace_id_tag_name is empty")
+ }
+ if trace.TimestampTagName == "" {
+ return errors.New("timestamp_tag_name is empty")
+ }
+ for i := range trace.Tags {
+ if trace.Tags[i].Name == "" {
+ return errors.New("trace tag name is empty")
+ }
+ if trace.Tags[i].Type == databasev1.TagType_TAG_TYPE_UNSPECIFIED {
+ return errors.New("trace tag type is unspecified")
+ }
+ }
+ return nil
+}
+
func tagFamily(tagFamilies []*databasev1.TagFamilySpec) error {
for i := range tagFamilies {
if tagFamilies[i].Name == "" {
diff --git a/banyand/metadata/client.go b/banyand/metadata/client.go
index f935f432..f192603c 100644
--- a/banyand/metadata/client.go
+++ b/banyand/metadata/client.go
@@ -243,6 +243,10 @@ func (s *clientService) MeasureRegistry() schema.Measure {
return s.schemaRegistry
}
+func (s *clientService) TraceRegistry() schema.Trace {
+ return s.schemaRegistry
+}
+
func (s *clientService) GroupRegistry() schema.Group {
return s.schemaRegistry
}
diff --git a/banyand/metadata/metadata.go b/banyand/metadata/metadata.go
index 14467c50..9a0024bc 100644
--- a/banyand/metadata/metadata.go
+++ b/banyand/metadata/metadata.go
@@ -45,6 +45,7 @@ type Repo interface {
IndexRuleRegistry() schema.IndexRule
IndexRuleBindingRegistry() schema.IndexRuleBinding
MeasureRegistry() schema.Measure
+ TraceRegistry() schema.Trace
GroupRegistry() schema.Group
TopNAggregationRegistry() schema.TopNAggregation
RegisterHandler(string, schema.Kind, schema.EventHandler)
diff --git a/banyand/metadata/schema/checker.go b/banyand/metadata/schema/checker.go
index 9dc140e2..7503360d 100644
--- a/banyand/metadata/schema/checker.go
+++ b/banyand/metadata/schema/checker.go
@@ -62,6 +62,13 @@ var CheckerMap = map[Kind]EqualityChecker{
protocmp.IgnoreFields(&commonv1.Metadata{}, "id", "create_revision", "mod_revision"),
protocmp.Transform())
},
+ KindTrace: func(a, b proto.Message) bool {
+ return cmp.Equal(a, b,
+ protocmp.IgnoreUnknown(),
+ protocmp.IgnoreFields(&databasev1.Trace{}, "updated_at"),
+ protocmp.IgnoreFields(&commonv1.Metadata{}, "id", "create_revision", "mod_revision"),
+ protocmp.Transform())
+ },
KindGroup: func(a, b proto.Message) bool {
return cmp.Equal(a, b,
protocmp.IgnoreUnknown(),
diff --git a/banyand/metadata/schema/etcd.go b/banyand/metadata/schema/etcd.go
index 73f8305a..a6372560 100644
--- a/banyand/metadata/schema/etcd.go
+++ b/banyand/metadata/schema/etcd.go
@@ -52,6 +52,7 @@ var (
_ IndexRuleBinding = (*etcdSchemaRegistry)(nil)
_ IndexRule = (*etcdSchemaRegistry)(nil)
_ Measure = (*etcdSchemaRegistry)(nil)
+ _ Trace = (*etcdSchemaRegistry)(nil)
_ Group = (*etcdSchemaRegistry)(nil)
errUnexpectedNumberOfEntities = errors.New("unexpected number of entities")
diff --git a/banyand/metadata/schema/kind.go b/banyand/metadata/schema/kind.go
index 3a955015..f164518e 100644
--- a/banyand/metadata/schema/kind.go
+++ b/banyand/metadata/schema/kind.go
@@ -36,15 +36,16 @@ const (
KindGroup Kind = 1 << iota
KindStream
KindMeasure
+ KindTrace
KindIndexRuleBinding
KindIndexRule
KindTopNAggregation
KindNode
KindProperty
- KindMask = KindGroup | KindStream | KindMeasure |
+ KindMask = KindGroup | KindStream | KindMeasure | KindTrace |
KindIndexRuleBinding | KindIndexRule |
KindTopNAggregation | KindNode | KindProperty
- KindSize = 8
+ KindSize = 9
)
func (k Kind) key() string {
@@ -55,6 +56,8 @@ func (k Kind) key() string {
return streamKeyPrefix
case KindMeasure:
return measureKeyPrefix
+ case KindTrace:
+ return traceKeyPrefix
case KindIndexRuleBinding:
return indexRuleBindingKeyPrefix
case KindIndexRule:
@@ -63,6 +66,8 @@ func (k Kind) key() string {
return topNAggregationKeyPrefix
case KindNode:
return nodeKeyPrefix
+ case KindProperty:
+ return propertyKeyPrefix
default:
return "unknown"
}
@@ -81,6 +86,8 @@ func (k Kind) Unmarshal(kv *mvccpb.KeyValue) (Metadata, error) {
m = &databasev1.Stream{}
case KindMeasure:
m = &databasev1.Measure{}
+ case KindTrace:
+ m = &databasev1.Trace{}
case KindIndexRuleBinding:
m = &databasev1.IndexRuleBinding{}
case KindIndexRule:
@@ -125,6 +132,8 @@ func (k Kind) String() string {
return "stream"
case KindMeasure:
return "measure"
+ case KindTrace:
+ return "trace"
case KindIndexRuleBinding:
return "indexRuleBinding"
case KindIndexRule:
@@ -133,6 +142,8 @@ func (k Kind) String() string {
return "topNAggregation"
case KindNode:
return "node"
+ case KindProperty:
+ return "property"
default:
return "unknown"
}
diff --git a/banyand/metadata/schema/schema.go b/banyand/metadata/schema/schema.go
index 6b98c05a..05e26f69 100644
--- a/banyand/metadata/schema/schema.go
+++ b/banyand/metadata/schema/schema.go
@@ -61,6 +61,7 @@ type Registry interface {
IndexRule
IndexRuleBinding
Measure
+ Trace
Group
TopNAggregation
Node
@@ -103,6 +104,11 @@ func (m Metadata) key() (string, error) {
Group: m.Group,
Name: m.Name,
}), nil
+ case KindTrace:
+ return formatTraceKey(&commonv1.Metadata{
+ Group: m.Group,
+ Name: m.Name,
+ }), nil
case KindIndexRule:
return formatIndexRuleKey(&commonv1.Metadata{
Group: m.Group,
@@ -180,6 +186,15 @@ type Measure interface {
TopNAggregations(ctx context.Context, metadata *commonv1.Metadata) ([]*databasev1.TopNAggregation, error)
}
+// Trace allows CRUD trace schemas in a group.
+type Trace interface {
+ GetTrace(ctx context.Context, metadata *commonv1.Metadata) (*databasev1.Trace, error)
+ ListTrace(ctx context.Context, opt ListOpt) ([]*databasev1.Trace, error)
+ CreateTrace(ctx context.Context, trace *databasev1.Trace) (int64, error)
+ UpdateTrace(ctx context.Context, trace *databasev1.Trace) (int64, error)
+ DeleteTrace(ctx context.Context, metadata *commonv1.Metadata) (bool, error)
+}
+
// Group allows CRUD groups which is namespaces of resources.
//
//go:generate mockgen -destination=group_mock.go -package=schema github.com/apache/skywalking-banyandb/banyand/metadata/schema Group
diff --git a/banyand/metadata/schema/trace.go b/banyand/metadata/schema/trace.go
new file mode 100644
index 00000000..74279407
--- /dev/null
+++ b/banyand/metadata/schema/trace.go
@@ -0,0 +1,83 @@
+// 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 schema
+
+import (
+ "context"
+
+ commonv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/common/v1"
+ databasev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/database/v1"
+)
+
+var traceKeyPrefix = "/traces/"
+
+func (e *etcdSchemaRegistry) GetTrace(ctx context.Context, metadata *commonv1.Metadata) (*databasev1.Trace, error) {
+ var trace databasev1.Trace
+ if err := e.get(ctx, formatTraceKey(metadata), &trace); err != nil {
+ return nil, err
+ }
+ return &trace, nil
+}
+
+func (e *etcdSchemaRegistry) ListTrace(ctx context.Context, opt ListOpt) ([]*databasev1.Trace, error) {
+ messages, err := e.listWithPrefix(ctx, listPrefixesForEntity(opt.Group, traceKeyPrefix), KindTrace)
+ if err != nil {
+ return nil, err
+ }
+ traces := make([]*databasev1.Trace, len(messages))
+ for i, message := range messages {
+ traces[i] = message.(*databasev1.Trace)
+ }
+ return traces, nil
+}
+
+func (e *etcdSchemaRegistry) CreateTrace(ctx context.Context, trace *databasev1.Trace) (int64, error) {
+ return e.create(ctx, Metadata{
+ TypeMeta: TypeMeta{
+ Kind: KindTrace,
+ Group: trace.GetMetadata().GetGroup(),
+ Name: trace.GetMetadata().GetName(),
+ },
+ Spec: trace,
+ })
+}
+
+func (e *etcdSchemaRegistry) UpdateTrace(ctx context.Context, trace *databasev1.Trace) (int64, error) {
+ return e.update(ctx, Metadata{
+ TypeMeta: TypeMeta{
+ Kind: KindTrace,
+ Group: trace.GetMetadata().GetGroup(),
+ Name: trace.GetMetadata().GetName(),
+ },
+ Spec: trace,
+ })
+}
+
+func (e *etcdSchemaRegistry) DeleteTrace(ctx context.Context, metadata *commonv1.Metadata) (bool, error) {
+ return e.delete(ctx, Metadata{
+ TypeMeta: TypeMeta{
+ Kind: KindTrace,
+ Group: metadata.GetGroup(),
+ Name: metadata.GetName(),
+ },
+ })
+}
+
+func formatTraceKey(metadata *commonv1.Metadata) string {
+ return formatKey(traceKeyPrefix, metadata)
+}
diff --git a/banyand/trace-plan.md b/banyand/trace-plan.md
deleted file mode 100644
index 5904076f..00000000
--- a/banyand/trace-plan.md
+++ /dev/null
@@ -1,264 +0,0 @@
-
-
-# Plan for Creating a New "trace" Package under `/banyand`
-
-Based on the analysis of the existing `stream` and `measure` packages, here's a comprehensive plan to create a new `trace` package following the established patterns:
-
-## 1. Package Structure
-
-```
-banyand/trace/
-├── svc_liaison.go # Liaison service implementation
-├── svc_standalone.go # Standalone service implementation
-├── metadata.go # Schema repository and suppliers
-├── trace.go # Core trace interface and implementation
-└── [other supporting files]
-```
-
-## 2. Core Interfaces and Types
-
-### 2.1 Service Interface
-```go
-// Service allows inspecting the trace data.
-type Service interface {
- run.PreRunner
- run.Config
- run.Service
- Query
-}
-
-// Query allows retrieving traces.
-type Query interface {
- LoadGroup(name string) (schema.Group, bool)
- Trace(metadata *commonv1.Metadata) (Trace, error)
- GetRemovalSegmentsTimeRange(group string) *timestamp.TimeRange
-}
-
-// Trace allows inspecting trace details.
-type Trace interface {
- GetSchema() *databasev1.Trace
- GetIndexRules() []*databasev1.IndexRule
- Query(ctx context.Context, opts model.TraceQueryOptions) (model.TraceQueryResult, error)
-}
-```
-
-### 2.2 Configuration Options
-```go
-type option struct {
- mergePolicy *mergePolicy
- protector protector.Memory
- tire2Client queue.Client
- seriesCacheMaxSize run.Bytes
- flushTimeout time.Duration
- elementIndexFlushTimeout time.Duration
-}
-```
-
-## 3. Service Implementations
-
-### 3.1 Liaison Service (`svc_liaison.go`)
-
-**Key Features:**
-- Implements `Service` interface
-- Role: `databasev1.Role_ROLE_LIAISON`
-- Uses `queueSupplier` for resource management
-- No pipeline listeners (as requested)
-- Handles distributed trace operations
-
-**Structure:**
-```go
-type liaison struct {
- pm protector.Memory
- metadata metadata.Repo
- pipeline queue.Server
- omr observability.MetricsRegistry
- lfs fs.FileSystem
- dataNodeSelector node.Selector
- l *logger.Logger
- schemaRepo schemaRepo
- dataPath string
- root string
- option option
- maxDiskUsagePercent int
-}
-```
-
-**Key Methods:**
-- `Trace(metadata *commonv1.Metadata) (Trace, error)`
-- `LoadGroup(name string) (resourceSchema.Group, bool)`
-- `GetRemovalSegmentsTimeRange(group string) *timestamp.TimeRange`
-- `FlagSet() *run.FlagSet`
-- `Validate() error`
-- `Name() string` (returns "trace")
-- `Role() databasev1.Role` (returns `ROLE_LIAISON`)
-- `PreRun(ctx context.Context) error`
-- `Serve() run.StopNotify`
-- `GracefulStop()`
-
-### 3.2 Standalone Service (`svc_standalone.go`)
-
-**Key Features:**
-- Implements `Service` interface
-- Role: `databasev1.Role_ROLE_DATA`
-- Uses `supplier` for resource management
-- No pipeline listeners (as requested)
-- Handles local trace operations
-
-**Structure:**
-```go
-type standalone struct {
- pm protector.Memory
- pipeline queue.Server
- localPipeline queue.Queue
- omr observability.MetricsRegistry
- lfs fs.FileSystem
- metadata metadata.Repo
- l *logger.Logger
- schemaRepo schemaRepo
- snapshotDir string
- root string
- dataPath string
- option option
- maxDiskUsagePercent int
- maxFileSnapshotNum int
-}
-```
-
-**Key Methods:**
-- `Trace(metadata *commonv1.Metadata) (Trace, error)`
-- `LoadGroup(name string) (resourceSchema.Group, bool)`
-- `GetRemovalSegmentsTimeRange(group string) *timestamp.TimeRange`
-- `FlagSet() *run.FlagSet`
-- `Validate() error`
-- `Name() string` (returns "trace")
-- `Role() databasev1.Role` (returns `ROLE_DATA`)
-- `PreRun(ctx context.Context) error`
-- `Serve() run.StopNotify`
-- `GracefulStop()`
-
-## 4. Schema Repository and Suppliers (`metadata.go`)
-
-### 4.1 Schema Repository
-```go
-type schemaRepo struct {
- resourceSchema.Repository
- l *logger.Logger
- metadata metadata.Repo
- path string
-}
-```
-
-### 4.2 Supplier (for Standalone)
-```go
-type supplier struct {
- metadata metadata.Repo
- omr observability.MetricsRegistry
- pm protector.Memory
- l *logger.Logger
- schemaRepo *schemaRepo
- nodeLabels map[string]string
- path string
- option option
-}
-```
-
-**Required Methods:**
-- `OpenResource(spec resourceSchema.Resource) (resourceSchema.IndexListener, error)`
-- `ResourceSchema(md *commonv1.Metadata) (resourceSchema.ResourceSchema, error)`
-- **Note: `OpenDB` method should NOT be implemented (as requested)**
-
-### 4.3 Queue Supplier (for Liaison)
-```go
-type queueSupplier struct {
- metadata metadata.Repo
- omr observability.MetricsRegistry
- pm protector.Memory
- traceDataNodeRegistry grpc.NodeRegistry
- l *logger.Logger
- schemaRepo *schemaRepo
- path string
- option option
-}
-```
-
-**Required Methods:**
-- `OpenResource(spec resourceSchema.Resource) (resourceSchema.IndexListener, error)`
-- `ResourceSchema(md *commonv1.Metadata) (resourceSchema.ResourceSchema, error)`
-- **Note: `OpenDB` method should NOT be implemented (as requested)**
-
-## 5. Core Trace Implementation (`trace.go`)
-
-```go
-type trace struct {
- pm protector.Memory
- indexSchema atomic.Value
- tsdb atomic.Value
- l *logger.Logger
- schema *databasev1.Trace
- schemaRepo *schemaRepo
- name string
- group string
-}
-
-type traceSpec struct {
- schema *databasev1.Trace
-}
-```
-
-## 6. Configuration and Flags
-
-### 6.1 Liaison Service Flags
-- `--trace-root-path` (default: "/tmp")
-- `--trace-data-path` (optional)
-- `--trace-flush-timeout` (default: 1 second)
-- `--trace-max-disk-usage-percent` (default: 95)
-
-### 6.2 Standalone Service Flags
-- `--trace-root-path` (default: "/tmp")
-- `--trace-data-path` (optional)
-- `--trace-flush-timeout` (default: 1 second)
-- `--trace-max-fan-out-size` (merge policy)
-- `--trace-series-cache-max-size` (default: 32MB)
-- `--trace-max-disk-usage-percent` (default: 95)
-- `--trace-max-file-snapshot-num` (default: 2)
-
-## 7. Error Handling
-
-Define common errors:
-```go
-var (
- errEmptyRootPath = errors.New("root path is empty")
- ErrTraceNotExist = errors.New("trace doesn't exist")
-)
-```
-
-## 8. Integration Points
-
-### 8.1 Metadata Integration
-- Use `metadata.TraceRegistry()` for trace schema operations
-- Implement proper schema watching and updates
-
-### 8.2 Storage Integration
-- Use `storage.TSDB` for time-series data storage
-- Implement proper segment management and TTL handling
-
-### 8.3 Metrics Integration
-- Implement trace-specific metrics
-- Use observability framework for monitoring
-
-## 9. Implementation Order
-
-1. **Core Interfaces** (`trace.go`)
-2. **Schema Repository** (`metadata.go`)
-3. **Standalone Service** (`svc_standalone.go`)
-4. **Liaison Service** (`svc_liaison.go`)
-5. **Supporting Files** (metrics, query, write handlers, etc.)
-
-## 10. Key Differences from Stream/Measure
-
-1. **No Pipeline Listeners**: Both services should not implement any pipeline listeners as requested
-2. **No OpenDB Implementation**: Both suppliers should implement all methods except `OpenDB`
-3. **Trace-Specific Schema**: Use `databasev1.Trace` instead of `databasev1.Stream` or `databasev1.Measure`
-4. **Trace-Specific Queries**: Implement trace-specific query patterns and result types
-
-This plan follows the established patterns from the stream and measure packages while adapting them specifically for trace data handling, ensuring consistency with the existing codebase architecture.
\ No newline at end of file
diff --git a/banyand/trace/metadata.go b/banyand/trace/metadata.go
new file mode 100644
index 00000000..1286decb
--- /dev/null
+++ b/banyand/trace/metadata.go
@@ -0,0 +1,184 @@
+// 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 trace
+
+import (
+ "context"
+ "time"
+
+ "github.com/pkg/errors"
+
+ commonv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/common/v1"
+ databasev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/database/v1"
+ "github.com/apache/skywalking-banyandb/banyand/liaison/grpc"
+ "github.com/apache/skywalking-banyandb/banyand/metadata"
+ "github.com/apache/skywalking-banyandb/banyand/observability"
+ "github.com/apache/skywalking-banyandb/banyand/protector"
+ "github.com/apache/skywalking-banyandb/pkg/logger"
+ resourceSchema "github.com/apache/skywalking-banyandb/pkg/schema"
+ "github.com/apache/skywalking-banyandb/pkg/timestamp"
+)
+
+var (
+ metadataScope = traceScope.SubScope("metadata")
+ ErrTraceNotFound = errors.New("trace not found")
+)
+
+// SchemaService allows querying schema information.
+type SchemaService interface {
+ Query
+ Close()
+}
+
+type schemaRepo struct {
+ resourceSchema.Repository
+ l *logger.Logger
+ metadata metadata.Repo
+ path string
+}
+
+func newSchemaRepo(path string, svc *standalone, nodeLabels map[string]string) schemaRepo {
+ sr := schemaRepo{
+ l: svc.l,
+ path: path,
+ metadata: svc.metadata,
+ Repository: resourceSchema.NewRepository(
+ svc.metadata,
+ svc.l,
+ newSupplier(path, svc, nodeLabels),
+ resourceSchema.NewMetrics(svc.omr.With(metadataScope)),
+ ),
+ }
+ sr.start()
+ return sr
+}
+
+func newLiaisonSchemaRepo(path string, svc *liaison, traceDataNodeRegistry grpc.NodeRegistry) schemaRepo {
+ sr := schemaRepo{
+ l: svc.l,
+ path: path,
+ metadata: svc.metadata,
+ Repository: resourceSchema.NewRepository(
+ svc.metadata,
+ svc.l,
+ newQueueSupplier(path, svc, traceDataNodeRegistry),
+ resourceSchema.NewMetrics(svc.omr.With(metadataScope)),
+ ),
+ }
+ sr.start()
+ return sr
+}
+
+func (sr *schemaRepo) start() {
+ sr.l.Info().Str("path", sr.path).Msg("starting trace metadata repository")
+}
+
+func (sr *schemaRepo) LoadGroup(name string) (resourceSchema.Group, bool) {
+ return sr.Repository.LoadGroup(name)
+}
+
+func (sr *schemaRepo) Trace(metadata *commonv1.Metadata) (*trace, bool) {
+ sm, ok := sr.Repository.LoadResource(metadata)
+ if !ok {
+ return nil, false
+ }
+ t, ok := sm.Delegated().(*trace)
+ return t, ok
+}
+
+func (sr *schemaRepo) GetRemovalSegmentsTimeRange(group string) *timestamp.TimeRange {
+ panic("not implemented")
+}
+
+// supplier is the supplier for standalone service
+type supplier struct {
+ metadata metadata.Repo
+ omr observability.MetricsRegistry
+ pm protector.Memory
+ l *logger.Logger
+ schemaRepo *schemaRepo
+ nodeLabels map[string]string
+ path string
+ option option
+}
+
+func newSupplier(path string, svc *standalone, nodeLabels map[string]string) *supplier {
+ return &supplier{
+ metadata: svc.metadata,
+ omr: svc.omr,
+ pm: svc.pm,
+ l: svc.l,
+ nodeLabels: nodeLabels,
+ path: path,
+ option: svc.option,
+ }
+}
+
+func (s *supplier) OpenResource(spec resourceSchema.Resource) (resourceSchema.IndexListener, error) {
+ traceSchema := spec.Schema().(*databasev1.Trace)
+ return openTrace(traceSchema, s.l, s.pm, s.schemaRepo), nil
+}
+
+func (s *supplier) ResourceSchema(md *commonv1.Metadata) (resourceSchema.ResourceSchema, error) {
+ ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+ defer cancel()
+ return s.metadata.TraceRegistry().GetTrace(ctx, md)
+}
+
+func (s *supplier) OpenDB(groupSchema *commonv1.Group) (resourceSchema.DB, error) {
+ panic("not implemented")
+}
+
+// queueSupplier is the supplier for liaison service
+type queueSupplier struct {
+ metadata metadata.Repo
+ omr observability.MetricsRegistry
+ pm protector.Memory
+ traceDataNodeRegistry grpc.NodeRegistry
+ l *logger.Logger
+ schemaRepo *schemaRepo
+ path string
+ option option
+}
+
+func newQueueSupplier(path string, svc *liaison, traceDataNodeRegistry grpc.NodeRegistry) *queueSupplier {
+ return &queueSupplier{
+ metadata: svc.metadata,
+ omr: svc.omr,
+ pm: svc.pm,
+ traceDataNodeRegistry: traceDataNodeRegistry,
+ l: svc.l,
+ path: path,
+ option: svc.option,
+ }
+}
+
+func (qs *queueSupplier) OpenResource(spec resourceSchema.Resource) (resourceSchema.IndexListener, error) {
+ traceSchema := spec.Schema().(*databasev1.Trace)
+ return openTrace(traceSchema, qs.l, qs.pm, qs.schemaRepo), nil
+}
+
+func (qs *queueSupplier) ResourceSchema(md *commonv1.Metadata) (resourceSchema.ResourceSchema, error) {
+ ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+ defer cancel()
+ return qs.metadata.TraceRegistry().GetTrace(ctx, md)
+}
+
+func (qs *queueSupplier) OpenDB(groupSchema *commonv1.Group) (resourceSchema.DB, error) {
+ panic("not implemented")
+}
diff --git a/banyand/trace/svc_liaison.go b/banyand/trace/svc_liaison.go
new file mode 100644
index 00000000..49912e67
--- /dev/null
+++ b/banyand/trace/svc_liaison.go
@@ -0,0 +1,190 @@
+// 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 trace
+
+import (
+ "context"
+ "path"
+
+ "github.com/pkg/errors"
+
+ commonv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/common/v1"
+ databasev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/database/v1"
+ "github.com/apache/skywalking-banyandb/banyand/liaison/grpc"
+ "github.com/apache/skywalking-banyandb/banyand/metadata"
+ "github.com/apache/skywalking-banyandb/banyand/observability"
+ "github.com/apache/skywalking-banyandb/banyand/protector"
+ "github.com/apache/skywalking-banyandb/banyand/queue"
+ "github.com/apache/skywalking-banyandb/pkg/fs"
+ "github.com/apache/skywalking-banyandb/pkg/logger"
+ "github.com/apache/skywalking-banyandb/pkg/node"
+ "github.com/apache/skywalking-banyandb/pkg/run"
+ resourceSchema "github.com/apache/skywalking-banyandb/pkg/schema"
+ "github.com/apache/skywalking-banyandb/pkg/timestamp"
+)
+
+type liaison struct {
+ pm protector.Memory
+ metadata metadata.Repo
+ pipeline queue.Server
+ omr observability.MetricsRegistry
+ lfs fs.FileSystem
+ dataNodeSelector node.Selector
+ l *logger.Logger
+ schemaRepo schemaRepo
+ dataPath string
+ root string
+ option option
+ maxDiskUsagePercent int
+}
+
+var _ Service = (*liaison)(nil)
+
+// LiaisonService returns a new liaison service.
+func LiaisonService(ctx context.Context) (Service, error) {
+ return &liaison{}, nil
+}
+
+func (l *liaison) FlagSet() *run.FlagSet {
+ fs := run.NewFlagSet("trace")
+ fs.StringVar(&l.root, "trace-root-path", "/tmp", "the root path for trace data")
+ fs.StringVar(&l.dataPath, "trace-data-path", "", "the path for trace data (optional)")
+ fs.DurationVar(&l.option.flushTimeout, "trace-flush-timeout", defaultFlushTimeout, "the timeout for trace data flush")
+ fs.IntVar(&l.maxDiskUsagePercent, "trace-max-disk-usage-percent", 95, "the maximum disk usage percentage")
+ // Additional liaison-specific flags can be added here
+ return fs
+}
+
+func (l *liaison) Validate() error {
+ if l.root == "" {
+ return errEmptyRootPath
+ }
+ return nil
+}
+
+func (l *liaison) Name() string {
+ return "trace"
+}
+
+func (l *liaison) Role() databasev1.Role {
+ return databasev1.Role_ROLE_LIAISON
+}
+
+func (l *liaison) PreRun(ctx context.Context) error {
+ l.l = logger.GetLogger("trace")
+
+ // Initialize metadata
+ if l.metadata == nil {
+ return errors.New("metadata repo is required")
+ }
+
+ // Initialize filesystem
+ if l.lfs == nil {
+ l.lfs = fs.NewLocalFileSystem()
+ }
+
+ // Initialize protector
+ if l.pm == nil {
+ return errors.New("memory protector is required")
+ }
+
+ // Initialize pipeline
+ if l.pipeline == nil {
+ return errors.New("pipeline is required")
+ }
+
+ // Set up data path
+ if l.dataPath == "" {
+ l.dataPath = path.Join(l.root, "trace-data")
+ }
+
+ // Initialize data node registry - this would be provided by the system
+ var traceDataNodeRegistry grpc.NodeRegistry
+
+ // Initialize schema repository
+ l.schemaRepo = newLiaisonSchemaRepo(l.dataPath, l, traceDataNodeRegistry)
+
+ l.l.Info().
+ Str("root", l.root).
+ Str("dataPath", l.dataPath).
+ Msg("trace liaison service initialized")
+
+ return nil
+}
+
+func (l *liaison) Serve() run.StopNotify {
+ // As specified in the plan, no pipeline listeners should be implemented
+ l.l.Info().Msg("trace liaison service started")
+
+ // Return a channel that never closes since this service runs indefinitely
+ stopCh := make(chan struct{})
+ return stopCh
+}
+
+func (l *liaison) GracefulStop() {
+ if l.schemaRepo.Repository != nil {
+ l.schemaRepo.Repository.Close()
+ }
+ l.l.Info().Msg("trace liaison service stopped")
+}
+
+func (l *liaison) LoadGroup(name string) (resourceSchema.Group, bool) {
+ return l.schemaRepo.LoadGroup(name)
+}
+
+func (l *liaison) Trace(metadata *commonv1.Metadata) (Trace, error) {
+ sm, ok := l.schemaRepo.Trace(metadata)
+ if !ok {
+ return nil, ErrTraceNotFound
+ }
+ return sm, nil
+}
+
+func (l *liaison) GetRemovalSegmentsTimeRange(group string) *timestamp.TimeRange {
+ return l.schemaRepo.GetRemovalSegmentsTimeRange(group)
+}
+
+// SetMetadata sets the metadata repository
+func (l *liaison) SetMetadata(metadata metadata.Repo) {
+ l.metadata = metadata
+}
+
+// SetObservabilityRegistry sets the observability metrics registry
+func (l *liaison) SetObservabilityRegistry(omr observability.MetricsRegistry) {
+ l.omr = omr
+}
+
+// SetProtector sets the memory protector
+func (l *liaison) SetProtector(pm protector.Memory) {
+ l.pm = pm
+}
+
+// SetPipeline sets the pipeline server
+func (l *liaison) SetPipeline(pipeline queue.Server) {
+ l.pipeline = pipeline
+}
+
+// SetFileSystem sets the file system
+func (l *liaison) SetFileSystem(lfs fs.FileSystem) {
+ l.lfs = lfs
+}
+
+// SetDataNodeSelector sets the data node selector
+func (l *liaison) SetDataNodeSelector(selector node.Selector) {
+ l.dataNodeSelector = selector
+}
diff --git a/banyand/trace/svc_standalone.go b/banyand/trace/svc_standalone.go
new file mode 100644
index 00000000..6be6451e
--- /dev/null
+++ b/banyand/trace/svc_standalone.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 trace
+
+import (
+ "context"
+ "path"
+ "path/filepath"
+
+ "github.com/pkg/errors"
+
+ commonv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/common/v1"
+ databasev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/database/v1"
+ "github.com/apache/skywalking-banyandb/banyand/metadata"
+ "github.com/apache/skywalking-banyandb/banyand/observability"
+ "github.com/apache/skywalking-banyandb/banyand/protector"
+ "github.com/apache/skywalking-banyandb/banyand/queue"
+ "github.com/apache/skywalking-banyandb/pkg/fs"
+ "github.com/apache/skywalking-banyandb/pkg/logger"
+ "github.com/apache/skywalking-banyandb/pkg/run"
+ resourceSchema "github.com/apache/skywalking-banyandb/pkg/schema"
+ "github.com/apache/skywalking-banyandb/pkg/timestamp"
+)
+
+var (
+ errEmptyRootPath = errors.New("root path is empty")
+ // ErrTraceNotExist denotes a trace doesn't exist in the metadata repo.
+ ErrTraceNotExist = errors.New("trace doesn't exist")
+)
+
+var _ Service = (*standalone)(nil)
+
+type standalone struct {
+ pm protector.Memory
+ pipeline queue.Server
+ localPipeline queue.Queue
+ omr observability.MetricsRegistry
+ lfs fs.FileSystem
+ metadata metadata.Repo
+ l *logger.Logger
+ schemaRepo schemaRepo
+ snapshotDir string
+ root string
+ dataPath string
+ option option
+ maxDiskUsagePercent int
+ maxFileSnapshotNum int
+}
+
+// StandaloneService returns a new standalone service.
+func StandaloneService(ctx context.Context) (Service, error) {
+ return &standalone{}, nil
+}
+
+func (s *standalone) FlagSet() *run.FlagSet {
+ fs := run.NewFlagSet("trace")
+ fs.StringVar(&s.root, "trace-root-path", "/tmp", "the root path for trace data")
+ fs.StringVar(&s.dataPath, "trace-data-path", "", "the path for trace data (optional)")
+ fs.DurationVar(&s.option.flushTimeout, "trace-flush-timeout", defaultFlushTimeout, "the timeout for trace data flush")
+ fs.IntVar(&s.maxDiskUsagePercent, "trace-max-disk-usage-percent", 95, "the maximum disk usage percentage")
+ fs.IntVar(&s.maxFileSnapshotNum, "trace-max-file-snapshot-num", 2, "the maximum number of file snapshots")
+ // Additional flags can be added here
+ return fs
+}
+
+func (s *standalone) Validate() error {
+ if s.root == "" {
+ return errEmptyRootPath
+ }
+ return nil
+}
+
+func (s *standalone) Name() string {
+ return "trace"
+}
+
+func (s *standalone) Role() databasev1.Role {
+ return databasev1.Role_ROLE_DATA
+}
+
+func (s *standalone) PreRun(ctx context.Context) error {
+ s.l = logger.GetLogger("trace")
+
+ // Initialize metadata
+ if s.metadata == nil {
+ return errors.New("metadata repo is required")
+ }
+
+ // Initialize filesystem
+ if s.lfs == nil {
+ s.lfs = fs.NewLocalFileSystem()
+ }
+
+ // Initialize protector
+ if s.pm == nil {
+ return errors.New("memory protector is required")
+ }
+
+ // Initialize pipeline
+ if s.pipeline == nil {
+ return errors.New("pipeline is required")
+ }
+
+ // Set up data path
+ if s.dataPath == "" {
+ s.dataPath = path.Join(s.root, "trace-data")
+ }
+
+ // Initialize schema repository
+ var nodeLabels map[string]string
+ s.schemaRepo = newSchemaRepo(s.dataPath, s, nodeLabels)
+
+ // Initialize snapshot directory
+ s.snapshotDir = filepath.Join(s.dataPath, "snapshot")
+
+ s.l.Info().
+ Str("root", s.root).
+ Str("dataPath", s.dataPath).
+ Str("snapshotDir", s.snapshotDir).
+ Msg("trace standalone service initialized")
+
+ return nil
+}
+
+func (s *standalone) Serve() run.StopNotify {
+ // As specified in the plan, no pipeline listeners should be implemented
+ s.l.Info().Msg("trace standalone service started")
+
+ // Return a channel that never closes since this service runs indefinitely
+ stopCh := make(chan struct{})
+ return stopCh
+}
+
+func (s *standalone) GracefulStop() {
+ if s.schemaRepo.Repository != nil {
+ s.schemaRepo.Repository.Close()
+ }
+ s.l.Info().Msg("trace standalone service stopped")
+}
+
+func (s *standalone) LoadGroup(name string) (resourceSchema.Group, bool) {
+ return s.schemaRepo.LoadGroup(name)
+}
+
+func (s *standalone) Trace(metadata *commonv1.Metadata) (Trace, error) {
+ sm, ok := s.schemaRepo.Trace(metadata)
+ if !ok {
+ return nil, ErrTraceNotFound
+ }
+ return sm, nil
+}
+
+func (s *standalone) GetRemovalSegmentsTimeRange(group string) *timestamp.TimeRange {
+ return s.schemaRepo.GetRemovalSegmentsTimeRange(group)
+}
+
+// SetMetadata sets the metadata repository
+func (s *standalone) SetMetadata(metadata metadata.Repo) {
+ s.metadata = metadata
+}
+
+// SetObservabilityRegistry sets the observability metrics registry
+func (s *standalone) SetObservabilityRegistry(omr observability.MetricsRegistry) {
+ s.omr = omr
+}
+
+// SetProtector sets the memory protector
+func (s *standalone) SetProtector(pm protector.Memory) {
+ s.pm = pm
+}
+
+// SetPipeline sets the pipeline server
+func (s *standalone) SetPipeline(pipeline queue.Server) {
+ s.pipeline = pipeline
+}
+
+// SetLocalPipeline sets the local pipeline queue
+func (s *standalone) SetLocalPipeline(localPipeline queue.Queue) {
+ s.localPipeline = localPipeline
+}
+
+// SetFileSystem sets the file system
+func (s *standalone) SetFileSystem(lfs fs.FileSystem) {
+ s.lfs = lfs
+}
diff --git a/banyand/trace/trace.go b/banyand/trace/trace.go
new file mode 100644
index 00000000..1f6d3e13
--- /dev/null
+++ b/banyand/trace/trace.go
@@ -0,0 +1,137 @@
+// 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 trace implements a trace-based storage which consists of trace data.
+// Traces are composed of spans and support querying by trace ID and various tags.
+package trace
+
+import (
+ "sync/atomic"
+ "time"
+
+ commonv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/common/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/protector"
+ "github.com/apache/skywalking-banyandb/banyand/queue"
+ "github.com/apache/skywalking-banyandb/pkg/logger"
+ "github.com/apache/skywalking-banyandb/pkg/run"
+ "github.com/apache/skywalking-banyandb/pkg/schema"
+ "github.com/apache/skywalking-banyandb/pkg/timestamp"
+)
+
+const (
+ defaultFlushTimeout = time.Second
+ defaultElementIndexFlushTimeout = time.Second
+)
+
+var traceScope = observability.RootScope.SubScope("trace")
+
+type option struct {
+ protector protector.Memory
+ tire2Client queue.Client
+ seriesCacheMaxSize run.Bytes
+ flushTimeout time.Duration
+ elementIndexFlushTimeout time.Duration
+}
+
+// Service allows inspecting the trace data.
+type Service interface {
+ run.PreRunner
+ run.Config
+ run.Service
+ Query
+}
+
+// Query allows retrieving traces.
+type Query interface {
+ LoadGroup(name string) (schema.Group, bool)
+ Trace(metadata *commonv1.Metadata) (Trace, error)
+ GetRemovalSegmentsTimeRange(group string) *timestamp.TimeRange
+}
+
+// Trace allows inspecting trace details.
+type Trace interface {
+ GetSchema() *databasev1.Trace
+ GetIndexRules() []*databasev1.IndexRule
+}
+
+type indexSchema struct {
+ tagMap map[string]*databasev1.TraceTagSpec
+ indexRuleLocators interface{} // Will be properly typed when trace partition support is added
+ indexRules []*databasev1.IndexRule
+}
+
+func (i *indexSchema) parse(schema *databasev1.Trace) {
+ // Note: This will need proper implementation when trace partition support is added
+ // For now, we'll just parse the tag map
+ i.tagMap = make(map[string]*databasev1.TraceTagSpec)
+ for _, tag := range schema.GetTags() {
+ i.tagMap[tag.GetName()] = tag
+ }
+}
+
+type trace struct {
+ pm protector.Memory
+ indexSchema atomic.Value
+ tsdb atomic.Value
+ l *logger.Logger
+ schema *databasev1.Trace
+ schemaRepo *schemaRepo
+ name string
+ group string
+}
+
+type traceSpec struct {
+ schema *databasev1.Trace
+}
+
+func (t *trace) GetSchema() *databasev1.Trace {
+ return t.schema
+}
+
+func (t *trace) GetIndexRules() []*databasev1.IndexRule {
+ if is := t.indexSchema.Load(); is != nil {
+ return is.(*indexSchema).indexRules
+ }
+ return nil
+}
+
+func (t *trace) OnIndexUpdate(index []*databasev1.IndexRule) {
+ var is indexSchema
+ is.indexRules = index
+ is.parse(t.schema)
+ t.indexSchema.Store(is)
+}
+
+func (t *trace) parseSpec() {
+ t.name, t.group = t.schema.GetMetadata().GetName(), t.schema.GetMetadata().GetGroup()
+ var is indexSchema
+ is.parse(t.schema)
+ t.indexSchema.Store(is)
+}
+
+func openTrace(schema *databasev1.Trace, l *logger.Logger, pm protector.Memory, schemaRepo *schemaRepo) *trace {
+ t := &trace{
+ schema: schema,
+ l: l,
+ pm: pm,
+ schemaRepo: schemaRepo,
+ }
+ t.parseSpec()
+ return t
+}