You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@skywalking.apache.org by li...@apache.org on 2021/06/29 11:33:08 UTC
[skywalking-satellite] 01/01: add meter pipe
This is an automated email from the ASF dual-hosted git repository.
liujiapeng pushed a commit to branch add-meter-pipe
in repository https://gitbox.apache.org/repos/asf/skywalking-satellite.git
commit fbc62502520fbec3f47fae17b8a0462df73c3138
Author: Evan <ev...@outlook.com>
AuthorDate: Tue Jun 29 19:32:37 2021 +0800
add meter pipe
---
configs/satellite_config.yaml | 24 ++++++-
....md => forwarder_nativemeter-grpc-forwarder.md} | 2 +-
docs/en/setup/plugins/plugin-list.md | 3 +-
.../plugins/receiver_grpc-nativemeter-receiver.md | 5 ++
docs/menu.yml | 6 +-
plugins/forwarder/forwarder_repository.go | 2 +-
.../nativecds/{sync_forwarder.go => forwarder.go} | 0
.../nativelog/{sync_forwarder.go => forwarder.go} | 0
.../{sync_forwarder.go => forwarder.go} | 0
.../sync_forwarder.go => nativemeter/forwarder.go} | 4 +-
.../{sync_forwarder.go => forwarder.go} | 0
.../{sync_forwarder.go => forwarder.go} | 0
.../nativelog/{sync_forwarder.go => forwarder.go} | 0
plugins/receiver/grpc/nativemeter/meter_service.go | 57 +++++++++++++++++
plugins/receiver/grpc/nativemeter/receiver.go | 61 ++++++++++++++++++
plugins/receiver/grpc/nativemeter/receiver_test.go | 74 ++++++++++++++++++++++
plugins/receiver/receiver_repository.go | 2 +
17 files changed, 232 insertions(+), 8 deletions(-)
diff --git a/configs/satellite_config.yaml b/configs/satellite_config.yaml
index 7b53e35..355a16c 100644
--- a/configs/satellite_config.yaml
+++ b/configs/satellite_config.yaml
@@ -180,4 +180,26 @@ pipes:
plugin_name: none-fallbacker
client_name: grpc-client
forwarders:
- - plugin_name: nativecds-grpc-forwarder
\ No newline at end of file
+ - plugin_name: nativecds-grpc-forwarder
+ - common_config:
+ pipe_name: meterpipe
+ gatherer:
+ server_name: "grpc-server"
+ receiver:
+ plugin_name: "grpc-nativemeter-receiver"
+ queue:
+ plugin_name: "memory-queue"
+ processor:
+ filters:
+ sender:
+ fallbacker:
+ plugin_name: none-fallbacker
+ # The time interval between two flush operations. And the time unit is millisecond.
+ flush_time: ${SATELLITE_TRACINGPIPE_SENDER_FLUSH_TIME:1000}
+ # The maximum buffer elements.
+ max_buffer_size: ${SATELLITE_TRACINGPIPE_SENDER_MAX_BUFFER_SIZE:200}
+ # The minimum flush elements.
+ min_flush_events: ${SATELLITE_TRACINGPIPE_SENDER_MIN_FLUSH_EVENTS:1}
+ client_name: grpc-client
+ forwarders:
+ - plugin_name: nativemeter-grpc-forwarder
\ No newline at end of file
diff --git a/docs/en/setup/plugins/forwarder_meter-grpc-forwarder.md b/docs/en/setup/plugins/forwarder_nativemeter-grpc-forwarder.md
similarity index 76%
rename from docs/en/setup/plugins/forwarder_meter-grpc-forwarder.md
rename to docs/en/setup/plugins/forwarder_nativemeter-grpc-forwarder.md
index 83005b0..3a8d083 100755
--- a/docs/en/setup/plugins/forwarder_meter-grpc-forwarder.md
+++ b/docs/en/setup/plugins/forwarder_nativemeter-grpc-forwarder.md
@@ -1,4 +1,4 @@
-# Forwarder/meter-grpc-forwarder
+# Forwarder/nativemeter-grpc-forwarder
## Description
This is a synchronization meter grpc forwarder with the SkyWalking meter protocol.
## DefaultConfig
diff --git a/docs/en/setup/plugins/plugin-list.md b/docs/en/setup/plugins/plugin-list.md
index b2f6bc7..1041d6e 100755
--- a/docs/en/setup/plugins/plugin-list.md
+++ b/docs/en/setup/plugins/plugin-list.md
@@ -9,11 +9,11 @@
- [prometheus-metrics-fetcher](./fetcher_prometheus-metrics-fetcher.md)
- Filter
- Forwarder
- - [meter-grpc-forwarder](./forwarder_meter-grpc-forwarder.md)
- [nativecds-grpc-forwarder](./forwarder_nativecds-grpc-forwarder.md)
- [nativelog-grpc-forwarder](./forwarder_nativelog-grpc-forwarder.md)
- [nativelog-kafka-forwarder](./forwarder_nativelog-kafka-forwarder.md)
- [nativemanagement-grpc-forwarder](./forwarder_nativemanagement-grpc-forwarder.md)
+ - [nativemeter-grpc-forwarder](./forwarder_nativemeter-grpc-forwarder.md)
- [nativeprofile-grpc-forwarder](./forwarder_nativeprofile-grpc-forwarder.md)
- [nativetracing-grpc-forwarder](./forwarder_nativetracing-grpc-forwarder.md)
- Parser
@@ -25,6 +25,7 @@
- [grpc-nativecds-receiver](./receiver_grpc-nativecds-receiver.md)
- [grpc-nativelog-receiver](./receiver_grpc-nativelog-receiver.md)
- [grpc-nativemanagement-receiver](./receiver_grpc-nativemanagement-receiver.md)
+ - [grpc-nativemeter-receiver](./receiver_grpc-nativemeter-receiver.md)
- [grpc-nativeprofile-receiver](./receiver_grpc-nativeprofile-receiver.md)
- [grpc-nativetracing-receiver](./receiver_grpc-nativetracing-receiver.md)
- [http-nativelog-receiver](./receiver_http-nativelog-receiver.md)
diff --git a/docs/en/setup/plugins/receiver_grpc-nativemeter-receiver.md b/docs/en/setup/plugins/receiver_grpc-nativemeter-receiver.md
new file mode 100755
index 0000000..8988fcd
--- /dev/null
+++ b/docs/en/setup/plugins/receiver_grpc-nativemeter-receiver.md
@@ -0,0 +1,5 @@
+# Receiver/grpc-nativemeter-receiver
+## Description
+This is a receiver for SkyWalking native meter format, which is defined at https://github.com/apache/skywalking-data-collect-protocol/blob/master/event/Event.proto.
+## DefaultConfig
+```yaml```
diff --git a/docs/menu.yml b/docs/menu.yml
index 9f7d2e9..cff2164 100644
--- a/docs/menu.yml
+++ b/docs/menu.yml
@@ -69,8 +69,6 @@ catalog:
path: /en/setup/plugins/fetcher_prometheus-metrics-fetcher
- name: forwarder
catalog:
- - name: meter-grpc-forwarder
- path: /en/setup/plugins/forwarder_meter-grpc-forwarder
- name: nativecds-grpc-forwarder
path: /en/setup/plugins/forwarder_nativecds-grpc-forwarder
- name: nativelog-grpc-forwarder
@@ -79,6 +77,8 @@ catalog:
path: /en/setup/plugins/forwarder_nativelog-kafka-forwarder
- name: nativemanagement-grpc-forwarder
path: /en/setup/plugins/forwarder_nativemanagement-grpc-forwarder
+ - name: nativemeter-grpc-forwarder
+ path: /en/setup/plugins/forwarder_nativemeter-grpc-forwarder
- name: nativeprofile-grpc-forwarder
path: /en/setup/plugins/forwarder_nativeprofile-grpc-forwarder
- name: nativetracing-grpc-forwarder
@@ -99,6 +99,8 @@ catalog:
path: /en/setup/plugins/receiver_grpc-nativelog-receiver
- name: grpc-nativemanagement-receiver
path: /en/setup/plugins/receiver_grpc-nativemanagement-receiver
+ - name: grpc-nativemeter-receiver
+ path: /en/setup/plugins/receiver_grpc-nativemeter-receiver
- name: grpc-nativeprofile-receiver
path: /en/setup/plugins/receiver_grpc-nativeprofile-receiver
- name: grpc-nativetracing-receiver
diff --git a/plugins/forwarder/forwarder_repository.go b/plugins/forwarder/forwarder_repository.go
index 3be46ba..ca02c2b 100644
--- a/plugins/forwarder/forwarder_repository.go
+++ b/plugins/forwarder/forwarder_repository.go
@@ -20,10 +20,10 @@ package forwarder
import (
"reflect"
- grpc_meter "github.com/apache/skywalking-satellite/plugins/forwarder/grpc/meter"
grpc_nativecds "github.com/apache/skywalking-satellite/plugins/forwarder/grpc/nativecds"
grpc_nativelog "github.com/apache/skywalking-satellite/plugins/forwarder/grpc/nativelog"
grpc_nativemanagement "github.com/apache/skywalking-satellite/plugins/forwarder/grpc/nativemanagement"
+ grpc_meter "github.com/apache/skywalking-satellite/plugins/forwarder/grpc/nativemeter"
grpc_nativeprofile "github.com/apache/skywalking-satellite/plugins/forwarder/grpc/nativeprofile"
grpc_nativetracing "github.com/apache/skywalking-satellite/plugins/forwarder/grpc/nativetracing"
kafka_nativelog "github.com/apache/skywalking-satellite/plugins/forwarder/kafka/nativelog"
diff --git a/plugins/forwarder/grpc/nativecds/sync_forwarder.go b/plugins/forwarder/grpc/nativecds/forwarder.go
similarity index 100%
rename from plugins/forwarder/grpc/nativecds/sync_forwarder.go
rename to plugins/forwarder/grpc/nativecds/forwarder.go
diff --git a/plugins/forwarder/grpc/nativelog/sync_forwarder.go b/plugins/forwarder/grpc/nativelog/forwarder.go
similarity index 100%
rename from plugins/forwarder/grpc/nativelog/sync_forwarder.go
rename to plugins/forwarder/grpc/nativelog/forwarder.go
diff --git a/plugins/forwarder/grpc/nativemanagement/sync_forwarder.go b/plugins/forwarder/grpc/nativemanagement/forwarder.go
similarity index 100%
rename from plugins/forwarder/grpc/nativemanagement/sync_forwarder.go
rename to plugins/forwarder/grpc/nativemanagement/forwarder.go
diff --git a/plugins/forwarder/grpc/meter/sync_forwarder.go b/plugins/forwarder/grpc/nativemeter/forwarder.go
similarity index 97%
rename from plugins/forwarder/grpc/meter/sync_forwarder.go
rename to plugins/forwarder/grpc/nativemeter/forwarder.go
index 5df933d..13ec551 100644
--- a/plugins/forwarder/grpc/meter/sync_forwarder.go
+++ b/plugins/forwarder/grpc/nativemeter/forwarder.go
@@ -15,7 +15,7 @@
// specific language governing permissions and limitations
// under the License.
-package meter
+package nativemeter
import (
"context"
@@ -33,7 +33,7 @@ import (
"github.com/apache/skywalking-satellite/internal/satellite/event"
)
-const Name = "meter-grpc-forwarder"
+const Name = "nativemeter-grpc-forwarder"
type Forwarder struct {
config.CommonFields
diff --git a/plugins/forwarder/grpc/nativeprofile/sync_forwarder.go b/plugins/forwarder/grpc/nativeprofile/forwarder.go
similarity index 100%
rename from plugins/forwarder/grpc/nativeprofile/sync_forwarder.go
rename to plugins/forwarder/grpc/nativeprofile/forwarder.go
diff --git a/plugins/forwarder/grpc/nativetracing/sync_forwarder.go b/plugins/forwarder/grpc/nativetracing/forwarder.go
similarity index 100%
rename from plugins/forwarder/grpc/nativetracing/sync_forwarder.go
rename to plugins/forwarder/grpc/nativetracing/forwarder.go
diff --git a/plugins/forwarder/kafka/nativelog/sync_forwarder.go b/plugins/forwarder/kafka/nativelog/forwarder.go
similarity index 100%
rename from plugins/forwarder/kafka/nativelog/sync_forwarder.go
rename to plugins/forwarder/kafka/nativelog/forwarder.go
diff --git a/plugins/receiver/grpc/nativemeter/meter_service.go b/plugins/receiver/grpc/nativemeter/meter_service.go
new file mode 100644
index 0000000..51415d3
--- /dev/null
+++ b/plugins/receiver/grpc/nativemeter/meter_service.go
@@ -0,0 +1,57 @@
+// 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 nativeevent
+
+import (
+ "io"
+ "time"
+
+ common "skywalking.apache.org/repo/goapi/collect/common/v3"
+ meter "skywalking.apache.org/repo/goapi/collect/language/agent/v3"
+ v1 "skywalking.apache.org/repo/goapi/satellite/data/v1"
+)
+
+const eventName = "grpc-nativemeter-event"
+
+type MeterService struct {
+ receiveChannel chan *v1.SniffData
+ meter.UnimplementedMeterReportServiceServer
+}
+
+func (m *MeterService) Collect(stream meter.MeterReportService_CollectServer) error {
+ for {
+ item, err := stream.Recv()
+ if err == io.EOF {
+ return stream.SendAndClose(&common.Commands{})
+ }
+ if err != nil {
+ return err
+ }
+ d := &v1.SniffData{
+ Name: eventName,
+ Timestamp: time.Now().UnixNano() / 1e6,
+ Meta: nil,
+ Type: v1.SniffType_MeterType,
+ Remote: true,
+ Data: &v1.SniffData_Meter{
+ Meter: item,
+ },
+ }
+ m.receiveChannel <- d
+ }
+}
diff --git a/plugins/receiver/grpc/nativemeter/receiver.go b/plugins/receiver/grpc/nativemeter/receiver.go
new file mode 100644
index 0000000..346a230
--- /dev/null
+++ b/plugins/receiver/grpc/nativemeter/receiver.go
@@ -0,0 +1,61 @@
+// 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 nativeevent
+
+import (
+ "github.com/apache/skywalking-satellite/internal/pkg/config"
+ module "github.com/apache/skywalking-satellite/internal/satellite/module/api"
+ "github.com/apache/skywalking-satellite/plugins/receiver/grpc"
+
+ meter "skywalking.apache.org/repo/goapi/collect/language/agent/v3"
+ v1 "skywalking.apache.org/repo/goapi/satellite/data/v1"
+)
+
+const Name = "grpc-nativemeter-receiver"
+
+type Receiver struct {
+ config.CommonFields
+ grpc.CommonGRPCReceiverFields
+ service *MeterService
+}
+
+func (r *Receiver) Name() string {
+ return Name
+}
+
+func (r *Receiver) Description() string {
+ return "This is a receiver for SkyWalking native meter format, " +
+ "which is defined at https://github.com/apache/skywalking-data-collect-protocol/blob/master/event/Event.proto."
+}
+
+func (r *Receiver) DefaultConfig() string {
+ return ""
+}
+
+func (r *Receiver) RegisterHandler(server interface{}) {
+ r.CommonGRPCReceiverFields = *grpc.InitCommonGRPCReceiverFields(server)
+ r.service = &MeterService{receiveChannel: r.OutputChannel}
+ meter.RegisterMeterReportServiceServer(r.Server, r.service)
+}
+
+func (r *Receiver) RegisterSyncInvoker(_ module.SyncInvoker) {
+}
+
+func (r *Receiver) Channel() <-chan *v1.SniffData {
+ return r.OutputChannel
+}
diff --git a/plugins/receiver/grpc/nativemeter/receiver_test.go b/plugins/receiver/grpc/nativemeter/receiver_test.go
new file mode 100644
index 0000000..91c6a2d
--- /dev/null
+++ b/plugins/receiver/grpc/nativemeter/receiver_test.go
@@ -0,0 +1,74 @@
+// 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 nativeevent
+
+import (
+ "context"
+ "strconv"
+ "testing"
+ "time"
+
+ "google.golang.org/grpc"
+
+ meter "skywalking.apache.org/repo/goapi/collect/language/agent/v3"
+ v1 "skywalking.apache.org/repo/goapi/satellite/data/v1"
+
+ _ "github.com/apache/skywalking-satellite/internal/satellite/test"
+ receiver_grpc "github.com/apache/skywalking-satellite/plugins/receiver/grpc"
+)
+
+func TestReceiver_RegisterHandler(t *testing.T) {
+ receiver_grpc.TestReceiver(new(Receiver), func(t *testing.T, sequence int, conn *grpc.ClientConn, ctx context.Context) string {
+ client := meter.NewMeterReportServiceClient(conn)
+ data := initData(sequence)
+ collect, err := client.Collect(ctx)
+ if err != nil {
+ t.Fatalf("cannot open the stream send mode: %v", err)
+ }
+ if err := collect.Send(data); err != nil {
+ t.Fatalf("cannot send the data to the server: %v", err)
+ }
+ if err := collect.CloseSend(); err != nil {
+ t.Fatalf("cannot close the stream mode: %v", err)
+ }
+ return data.String()
+ }, func(data *v1.SniffData) string {
+ return data.GetMeter().String()
+ }, t)
+}
+
+func initData(sequence int) *meter.MeterData {
+ seq := strconv.Itoa(sequence)
+ return &meter.MeterData{
+ Service: "demo-service" + seq,
+ ServiceInstance: "demo-instance" + seq,
+ Timestamp: time.Now().Unix() / 1e6,
+ Metric: &meter.MeterData_SingleValue{
+ SingleValue: &meter.MeterSingleValue{
+ Name: "name" + seq,
+ Value: float64(sequence),
+ Labels: []*meter.Label{
+ {
+ Name: "label-name" + seq,
+ Value: "label-value" + seq,
+ },
+ },
+ },
+ },
+ }
+}
diff --git a/plugins/receiver/receiver_repository.go b/plugins/receiver/receiver_repository.go
index d6ed000..5cbc52c 100644
--- a/plugins/receiver/receiver_repository.go
+++ b/plugins/receiver/receiver_repository.go
@@ -25,6 +25,7 @@ import (
grpcnativecds "github.com/apache/skywalking-satellite/plugins/receiver/grpc/nativecds"
grpcnavtivelog "github.com/apache/skywalking-satellite/plugins/receiver/grpc/nativelog"
grpcnativemanagement "github.com/apache/skywalking-satellite/plugins/receiver/grpc/nativemanagement"
+ grpcnativemeter "github.com/apache/skywalking-satellite/plugins/receiver/grpc/nativemeter"
grpcnativeprofile "github.com/apache/skywalking-satellite/plugins/receiver/grpc/nativeprofile"
grpcnativetracing "github.com/apache/skywalking-satellite/plugins/receiver/grpc/nativetracing"
httpnavtivelog "github.com/apache/skywalking-satellite/plugins/receiver/http/nativcelog"
@@ -41,6 +42,7 @@ func RegisterReceiverPlugins() {
new(grpcnativeprofile.Receiver),
new(grpcnativecds.Receiver),
new(httpnavtivelog.Receiver),
+ new(grpcnativemeter.Receiver),
}
for _, receiver := range receivers {
plugin.RegisterPlugin(receiver)