| // 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 data contains data transmission topics. |
| package data |
| |
| import ( |
| "google.golang.org/protobuf/proto" |
| |
| databasev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/database/v1" |
| measurev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/measure/v1" |
| propertyv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/property/v1" |
| streamv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/stream/v1" |
| tracev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/trace/v1" |
| "github.com/apache/skywalking-banyandb/pkg/bus" |
| ) |
| |
| var ( |
| // TopicMap is the map of topic name to topic. |
| TopicMap = map[string]bus.Topic{ |
| TopicStreamWrite.String(): TopicStreamWrite, |
| TopicStreamQuery.String(): TopicStreamQuery, |
| TopicMeasureWrite.String(): TopicMeasureWrite, |
| TopicMeasureQuery.String(): TopicMeasureQuery, |
| TopicInternalMeasureQuery.String(): TopicInternalMeasureQuery, |
| TopicTopNQuery.String(): TopicTopNQuery, |
| TopicPropertyDelete.String(): TopicPropertyDelete, |
| TopicPropertyQuery.String(): TopicPropertyQuery, |
| TopicPropertyUpdate.String(): TopicPropertyUpdate, |
| TopicStreamPartSync.String(): TopicStreamPartSync, |
| TopicMeasurePartSync.String(): TopicMeasurePartSync, |
| TopicMeasureSeriesIndexInsert.String(): TopicMeasureSeriesIndexInsert, |
| TopicMeasureSeriesIndexUpdate.String(): TopicMeasureSeriesIndexUpdate, |
| TopicMeasureSeriesSync.String(): TopicMeasureSeriesSync, |
| TopicPropertyRepair.String(): TopicPropertyRepair, |
| TopicStreamSeriesIndexWrite.String(): TopicStreamSeriesIndexWrite, |
| TopicStreamLocalIndexWrite.String(): TopicStreamLocalIndexWrite, |
| TopicStreamSeriesSync.String(): TopicStreamSeriesSync, |
| TopicStreamElementIndexSync.String(): TopicStreamElementIndexSync, |
| TopicTraceWrite.String(): TopicTraceWrite, |
| TopicTraceQuery.String(): TopicTraceQuery, |
| TopicTracePartSync.String(): TopicTracePartSync, |
| TopicTraceSeriesSync.String(): TopicTraceSeriesSync, |
| TopicTraceSidxSeriesWrite.String(): TopicTraceSidxSeriesWrite, |
| TopicMeasureCollectDataInfo.String(): TopicMeasureCollectDataInfo, |
| TopicMeasureCollectLiaisonInfo.String(): TopicMeasureCollectLiaisonInfo, |
| TopicStreamCollectDataInfo.String(): TopicStreamCollectDataInfo, |
| TopicStreamCollectLiaisonInfo.String(): TopicStreamCollectLiaisonInfo, |
| TopicTraceCollectDataInfo.String(): TopicTraceCollectDataInfo, |
| TopicTraceCollectLiaisonInfo.String(): TopicTraceCollectLiaisonInfo, |
| TopicMeasureDropGroup.String(): TopicMeasureDropGroup, |
| TopicStreamDropGroup.String(): TopicStreamDropGroup, |
| TopicTraceDropGroup.String(): TopicTraceDropGroup, |
| } |
| |
| // TopicRequestMap is the map of topic name to request message. |
| // nolint: exhaustruct |
| TopicRequestMap = map[bus.Topic]func() proto.Message{ |
| TopicStreamWrite: func() proto.Message { |
| return &streamv1.InternalWriteRequest{} |
| }, |
| TopicStreamQuery: func() proto.Message { |
| return &streamv1.QueryRequest{} |
| }, |
| TopicMeasureWrite: func() proto.Message { |
| return &measurev1.InternalWriteRequest{} |
| }, |
| TopicMeasureQuery: func() proto.Message { |
| return &measurev1.QueryRequest{} |
| }, |
| TopicInternalMeasureQuery: func() proto.Message { |
| return &measurev1.InternalQueryRequest{} |
| }, |
| TopicTopNQuery: func() proto.Message { |
| return &measurev1.TopNRequest{} |
| }, |
| TopicPropertyUpdate: func() proto.Message { |
| return &propertyv1.InternalUpdateRequest{} |
| }, |
| TopicPropertyQuery: func() proto.Message { |
| return &propertyv1.QueryRequest{} |
| }, |
| TopicPropertyDelete: func() proto.Message { |
| return &propertyv1.InternalDeleteRequest{} |
| }, |
| TopicStreamPartSync: func() proto.Message { |
| return nil |
| }, |
| TopicMeasurePartSync: func() proto.Message { |
| return nil |
| }, |
| TopicMeasureSeriesIndexInsert: func() proto.Message { |
| return nil |
| }, |
| TopicMeasureSeriesIndexUpdate: func() proto.Message { |
| return nil |
| }, |
| TopicMeasureSeriesSync: func() proto.Message { |
| return nil |
| }, |
| TopicPropertyRepair: func() proto.Message { |
| return &propertyv1.InternalRepairRequest{} |
| }, |
| TopicStreamSeriesIndexWrite: func() proto.Message { |
| return nil |
| }, |
| TopicStreamLocalIndexWrite: func() proto.Message { |
| return nil |
| }, |
| TopicStreamSeriesSync: func() proto.Message { |
| return nil |
| }, |
| TopicStreamElementIndexSync: func() proto.Message { |
| return nil |
| }, |
| TopicTraceWrite: func() proto.Message { |
| return &tracev1.InternalWriteRequest{} |
| }, |
| TopicTraceQuery: func() proto.Message { |
| return &tracev1.QueryRequest{} |
| }, |
| TopicTracePartSync: func() proto.Message { |
| return nil |
| }, |
| TopicTraceSeriesSync: func() proto.Message { |
| return nil |
| }, |
| TopicTraceSidxSeriesWrite: func() proto.Message { |
| return nil |
| }, |
| TopicMeasureCollectDataInfo: func() proto.Message { |
| return &databasev1.GroupRegistryServiceInspectRequest{} |
| }, |
| TopicMeasureCollectLiaisonInfo: func() proto.Message { |
| return &databasev1.GroupRegistryServiceInspectRequest{} |
| }, |
| TopicStreamCollectDataInfo: func() proto.Message { |
| return &databasev1.GroupRegistryServiceInspectRequest{} |
| }, |
| TopicStreamCollectLiaisonInfo: func() proto.Message { |
| return &databasev1.GroupRegistryServiceInspectRequest{} |
| }, |
| TopicTraceCollectDataInfo: func() proto.Message { |
| return &databasev1.GroupRegistryServiceInspectRequest{} |
| }, |
| TopicTraceCollectLiaisonInfo: func() proto.Message { |
| return &databasev1.GroupRegistryServiceInspectRequest{} |
| }, |
| TopicMeasureDropGroup: func() proto.Message { |
| return &databasev1.GroupRegistryServiceDeleteRequest{} |
| }, |
| TopicStreamDropGroup: func() proto.Message { |
| return &databasev1.GroupRegistryServiceDeleteRequest{} |
| }, |
| TopicTraceDropGroup: func() proto.Message { |
| return &databasev1.GroupRegistryServiceDeleteRequest{} |
| }, |
| } |
| |
| // TopicResponseMap is the map of topic name to per-topic response codec. |
| // Every proto-bodied topic is wrapped in a behavior-preserving ProtoCodec |
| // (byte-identical to the pre-G9f.0 proto.Marshal/proto.Unmarshal path). |
| // TopicInternalMeasureQuery uses a wire-mode-dispatching codec: flag-on |
| // processes encode/decode the raw vec columnar frame via RawFrameCodec, |
| // flag-off processes keep the proto body via ProtoCodec (one static |
| // supplier serves both modes on the same topic — topic-AND-process-wire- |
| // mode selection, G9f spec G9f.0). |
| // nolint: exhaustruct |
| TopicResponseMap = map[bus.Topic]ResponseCodec{ |
| TopicStreamQuery: NewProtoCodec(func() proto.Message { |
| return &streamv1.QueryResponse{} |
| }), |
| TopicMeasureQuery: NewProtoCodec(func() proto.Message { |
| return &measurev1.QueryResponse{} |
| }), |
| TopicInternalMeasureQuery: &measureQueryResponseCodec{ |
| proto: NewProtoCodec(func() proto.Message { |
| return &measurev1.InternalQueryResponse{} |
| }), |
| raw: NewRawFrameCodec(), |
| }, |
| TopicTopNQuery: NewProtoCodec(func() proto.Message { |
| return &measurev1.TopNResponse{} |
| }), |
| TopicPropertyQuery: NewProtoCodec(func() proto.Message { |
| return &propertyv1.InternalQueryResponse{} |
| }), |
| TopicPropertyDelete: NewProtoCodec(func() proto.Message { |
| return &propertyv1.DeleteResponse{} |
| }), |
| TopicPropertyUpdate: NewProtoCodec(func() proto.Message { |
| return &propertyv1.ApplyResponse{} |
| }), |
| TopicPropertyRepair: NewProtoCodec(func() proto.Message { |
| return &propertyv1.InternalRepairResponse{} |
| }), |
| TopicTraceQuery: &traceQueryResponseCodec{ |
| proto: NewProtoCodec(func() proto.Message { |
| return &tracev1.InternalQueryResponse{} |
| }), |
| raw: NewRawFrameCodec(), |
| }, |
| TopicMeasureCollectDataInfo: NewProtoCodec(func() proto.Message { |
| return &databasev1.DataInfo{} |
| }), |
| TopicStreamCollectDataInfo: NewProtoCodec(func() proto.Message { |
| return &databasev1.DataInfo{} |
| }), |
| TopicTraceCollectDataInfo: NewProtoCodec(func() proto.Message { |
| return &databasev1.DataInfo{} |
| }), |
| TopicMeasureCollectLiaisonInfo: NewProtoCodec(func() proto.Message { |
| return &databasev1.LiaisonInfo{} |
| }), |
| TopicStreamCollectLiaisonInfo: NewProtoCodec(func() proto.Message { |
| return &databasev1.LiaisonInfo{} |
| }), |
| TopicTraceCollectLiaisonInfo: NewProtoCodec(func() proto.Message { |
| return &databasev1.LiaisonInfo{} |
| }), |
| TopicMeasureDropGroup: NewProtoCodec(func() proto.Message { |
| return &databasev1.GroupRegistryServiceDeleteRequest{} |
| }), |
| TopicStreamDropGroup: NewProtoCodec(func() proto.Message { |
| return &databasev1.GroupRegistryServiceDeleteRequest{} |
| }), |
| TopicTraceDropGroup: NewProtoCodec(func() proto.Message { |
| return &databasev1.GroupRegistryServiceDeleteRequest{} |
| }), |
| } |
| |
| // TopicCommon is the common topic for data transmission. |
| TopicCommon = bus.Topic{} |
| ) |
| |
| const ( |
| // OperationFileSyncValue is the operation label for file-sync (chunked sync) operations. |
| OperationFileSyncValue = "file-sync" |
| // OperationBatchWriteValue is the operation label for batch-write operations. |
| OperationBatchWriteValue = "batch-write" |
| // OperationQueryValue is the operation label for query operations. |
| OperationQueryValue = "query" |
| // OperationControlValue is the operation label for control operations. |
| OperationControlValue = "control" |
| ) |
| |
| var operationMap = map[bus.Topic]string{ |
| TopicStreamPartSync: OperationFileSyncValue, |
| TopicMeasurePartSync: OperationFileSyncValue, |
| TopicTracePartSync: OperationFileSyncValue, |
| TopicStreamSeriesSync: OperationFileSyncValue, |
| TopicMeasureSeriesSync: OperationFileSyncValue, |
| TopicTraceSeriesSync: OperationFileSyncValue, |
| TopicStreamElementIndexSync: OperationFileSyncValue, |
| TopicStreamWrite: OperationBatchWriteValue, |
| TopicMeasureWrite: OperationBatchWriteValue, |
| TopicTraceWrite: OperationBatchWriteValue, |
| TopicMeasureSeriesIndexInsert: OperationBatchWriteValue, |
| TopicMeasureSeriesIndexUpdate: OperationBatchWriteValue, |
| TopicStreamSeriesIndexWrite: OperationBatchWriteValue, |
| TopicStreamLocalIndexWrite: OperationBatchWriteValue, |
| TopicTraceSidxSeriesWrite: OperationBatchWriteValue, |
| TopicStreamQuery: OperationQueryValue, |
| TopicMeasureQuery: OperationQueryValue, |
| TopicInternalMeasureQuery: OperationQueryValue, |
| TopicTopNQuery: OperationQueryValue, |
| TopicTraceQuery: OperationQueryValue, |
| TopicPropertyUpdate: OperationControlValue, |
| TopicPropertyDelete: OperationControlValue, |
| TopicPropertyQuery: OperationControlValue, |
| TopicPropertyRepair: OperationControlValue, |
| TopicMeasureCollectDataInfo: OperationControlValue, |
| TopicMeasureCollectLiaisonInfo: OperationControlValue, |
| TopicStreamCollectDataInfo: OperationControlValue, |
| TopicStreamCollectLiaisonInfo: OperationControlValue, |
| TopicTraceCollectDataInfo: OperationControlValue, |
| TopicTraceCollectLiaisonInfo: OperationControlValue, |
| TopicMeasureDropGroup: OperationControlValue, |
| TopicStreamDropGroup: OperationControlValue, |
| TopicTraceDropGroup: OperationControlValue, |
| } |
| |
| // OperationOf returns the operation label string for the given bus topic. |
| func OperationOf(topic bus.Topic) string { |
| if op, ok := operationMap[topic]; ok { |
| return op |
| } |
| return OperationControlValue |
| } |
| |
| // GroupFromMessageData extracts the business group string from the pre-marshal request proto (pub side). |
| func GroupFromMessageData(_ bus.Topic, payload any) string { |
| switch v := payload.(type) { |
| case *measurev1.InternalWriteRequest: |
| return groupFromWriteMetadata(v.GetRequest().GetMetadata()) |
| case *streamv1.InternalWriteRequest: |
| return groupFromWriteMetadata(v.GetRequest().GetMetadata()) |
| case *tracev1.InternalWriteRequest: |
| return groupFromWriteMetadata(v.GetRequest().GetMetadata()) |
| case *measurev1.QueryRequest: |
| return firstGroup(v.GetGroups()) |
| case *measurev1.InternalQueryRequest: |
| return firstGroup(v.GetRequest().GetGroups()) |
| case *measurev1.TopNRequest: |
| return firstGroup(v.GetGroups()) |
| case *streamv1.QueryRequest: |
| return firstGroup(v.GetGroups()) |
| case *tracev1.QueryRequest: |
| return firstGroup(v.GetGroups()) |
| case *propertyv1.InternalUpdateRequest: |
| return groupFromProperty(v.GetProperty()) |
| case *propertyv1.InternalRepairRequest: |
| return groupFromProperty(v.GetProperty()) |
| case *propertyv1.QueryRequest: |
| return firstGroup(v.GetGroups()) |
| case *databasev1.GroupRegistryServiceDeleteRequest: |
| return v.GetGroup() |
| } |
| return "" |
| } |
| |
| type groupGetter interface { |
| GetGroup() string |
| } |
| |
| func groupFromWriteMetadata(md groupGetter) string { |
| if md == nil { |
| return "" |
| } |
| return md.GetGroup() |
| } |
| |
| func groupFromProperty(p *propertyv1.Property) string { |
| if p == nil { |
| return "" |
| } |
| if md := p.GetMetadata(); md != nil { |
| return md.GetGroup() |
| } |
| return "" |
| } |
| |
| func firstGroup(groups []string) string { |
| if len(groups) > 0 { |
| return groups[0] |
| } |
| return "" |
| } |
