You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@skywalking.apache.org by wu...@apache.org on 2021/10/18 07:31:34 UTC
[skywalking-satellite] branch main updated: Support transmit Envoy
Metrics v2/v3 (#79)
This is an automated email from the ASF dual-hosted git repository.
wusheng pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/skywalking-satellite.git
The following commit(s) were added to refs/heads/main by this push:
new baddab2 Support transmit Envoy Metrics v2/v3 (#79)
baddab2 is described below
commit baddab2d58f2b87df59f475dc4b45b36fa0a2825
Author: mrproliu <74...@qq.com>
AuthorDate: Mon Oct 18 15:31:27 2021 +0800
Support transmit Envoy Metrics v2/v3 (#79)
---
.github/workflows/e2e.yaml | 2 +-
CHANGES.md | 1 +
configs/satellite_config.yaml | 52 ++++++++++
.../forwarder_envoy-metrics-v2-grpc-forwarder.md | 9 ++
.../forwarder_envoy-metrics-v3-grpc-forwarder.md | 9 ++
docs/en/setup/plugins/plugin-list.md | 4 +
.../receiver_grpc-envoy-metrics-v2-receiver.md | 18 ++++
.../receiver_grpc-envoy-metrics-v3-receiver.md | 18 ++++
docs/menu.yml | 8 ++
plugins/forwarder/forwarder_repository.go | 4 +
plugins/forwarder/grpc/envoymetricsv2/forwarder.go | 106 +++++++++++++++++++++
plugins/forwarder/grpc/envoymetricsv3/forwarder.go | 106 +++++++++++++++++++++
.../receiver/grpc/envoymetricsv2/als_service.go | 92 ++++++++++++++++++
plugins/receiver/grpc/envoymetricsv2/receiver.go | 77 +++++++++++++++
.../receiver/grpc/envoymetricsv2/receiver_test.go | 67 +++++++++++++
.../receiver/grpc/envoymetricsv3/als_service.go | 92 ++++++++++++++++++
plugins/receiver/grpc/envoymetricsv3/receiver.go | 78 +++++++++++++++
.../receiver/grpc/envoymetricsv3/receiver_test.go | 66 +++++++++++++
plugins/receiver/receiver_repository.go | 4 +
test/e2e/case/envoy/e2e.yaml | 7 +-
test/e2e/case/envoy/envoy-v1.16.2.yaml | 7 ++
test/e2e/case/envoy/envoy-v1.19.0.yaml | 9 ++
test/e2e/case/envoy/expected/metrics-has-value.yml | 19 ++++
23 files changed, 852 insertions(+), 3 deletions(-)
diff --git a/.github/workflows/e2e.yaml b/.github/workflows/e2e.yaml
index 397cf59..ec46b75 100644
--- a/.github/workflows/e2e.yaml
+++ b/.github/workflows/e2e.yaml
@@ -38,7 +38,7 @@ jobs:
with:
e2e-file: test/e2e/case/native-protocols/e2e.yaml
EnvoyE2E:
- name: Envoy ALS
+ name: Envoy ALS/Metrics
runs-on: ubuntu-latest
strategy:
matrix:
diff --git a/CHANGES.md b/CHANGES.md
index dca13ba..35fbbb7 100644
--- a/CHANGES.md
+++ b/CHANGES.md
@@ -8,6 +8,7 @@ Release Notes.
* Support load-balance GRPC client with the static server list.
* Support load-balance GRPC client with the Kubernetes selector.
* Support transmit Envoy ALS v2/v3 protocol.
+* Support transmit Envoy Metrics v2/v3 protocol.
#### Bug Fixes
* Fix errors when converting meter data from histogram and summary.[#75](https://github.com/apache/skywalking-satellite/pull/75)
diff --git a/configs/satellite_config.yaml b/configs/satellite_config.yaml
index a58cd53..c7ec8e8 100644
--- a/configs/satellite_config.yaml
+++ b/configs/satellite_config.yaml
@@ -299,3 +299,55 @@ pipes:
client_name: grpc-client
forwarders:
- plugin_name: envoy-als-v3-grpc-forwarder
+ - common_config:
+ pipe_name: envoy-metrics-v2
+ gatherer:
+ server_name: "grpc-server"
+ receiver:
+ plugin_name: "grpc-envoy-metrics-v2-receiver"
+ # The time interval between two flush operations. And the time unit is millisecond.
+ flush_time: ${SATELLITE_ENVOY_METRICS_V2_PIPE_RECEIVER_FLUSH_TIME:1000}
+ # The max cache count when receive the message
+ limit_count: ${SATELLITE_ENVOY_METRICS_V2_PIPE_RECEIVER_LIMIT_COUNT:500}
+ 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_ENVOY_METRICS_V2_PIPE_SENDER_FLUSH_TIME:1000}
+ # The maximum buffer elements.
+ max_buffer_size: ${SATELLITE_ENVOY_METRICS_V2_PIPE_SENDER_MAX_BUFFER_SIZE:200}
+ # The minimum flush elements.
+ min_flush_events: ${SATELLITE_ENVOY_METRICS_V2_PIPE_SENDER_MIN_FLUSH_EVENTS:1}
+ client_name: grpc-client
+ forwarders:
+ - plugin_name: envoy-metrics-v2-grpc-forwarder
+ - common_config:
+ pipe_name: envoy-metrics-v3
+ gatherer:
+ server_name: "grpc-server"
+ receiver:
+ plugin_name: "grpc-envoy-metrics-v3-receiver"
+ # The time interval between two flush operations. And the time unit is millisecond.
+ flush_time: ${SATELLITE_ENVOY_METRICS_V3_PIPE_RECEIVER_FLUSH_TIME:1000}
+ # The max cache count when receive the message
+ limit_count: ${SATELLITE_ENVOY_METRICS_V3_PIPE_RECEIVER_LIMIT_COUNT:500}
+ 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_ENVOY_METRICS_V3_PIPE_SENDER_FLUSH_TIME:1000}
+ # The maximum buffer elements.
+ max_buffer_size: ${SATELLITE_ENVOY_METRICS_V3_PIPE_SENDER_MAX_BUFFER_SIZE:200}
+ # The minimum flush elements.
+ min_flush_events: ${SATELLITE_ENVOY_METRICS_V3_PIPE_SENDER_MIN_FLUSH_EVENTS:1}
+ client_name: grpc-client
+ forwarders:
+ - plugin_name: envoy-metrics-v3-grpc-forwarder
diff --git a/docs/en/setup/plugins/forwarder_envoy-metrics-v2-grpc-forwarder.md b/docs/en/setup/plugins/forwarder_envoy-metrics-v2-grpc-forwarder.md
new file mode 100755
index 0000000..6989e42
--- /dev/null
+++ b/docs/en/setup/plugins/forwarder_envoy-metrics-v2-grpc-forwarder.md
@@ -0,0 +1,9 @@
+# Forwarder/envoy-metrics-v2-grpc-forwarder
+## Description
+This is a synchronization Metrics v2 grpc forwarder with the Envoy metrics protocol.
+## DefaultConfig
+```yaml```
+## Configuration
+|Name|Type|Description|
+|----|----|-----------|
+
diff --git a/docs/en/setup/plugins/forwarder_envoy-metrics-v3-grpc-forwarder.md b/docs/en/setup/plugins/forwarder_envoy-metrics-v3-grpc-forwarder.md
new file mode 100755
index 0000000..41de497
--- /dev/null
+++ b/docs/en/setup/plugins/forwarder_envoy-metrics-v3-grpc-forwarder.md
@@ -0,0 +1,9 @@
+# Forwarder/envoy-metrics-v3-grpc-forwarder
+## Description
+This is a synchronization Metrics v3 grpc forwarder with the Envoy metrics protocol.
+## DefaultConfig
+```yaml```
+## Configuration
+|Name|Type|Description|
+|----|----|-----------|
+
diff --git a/docs/en/setup/plugins/plugin-list.md b/docs/en/setup/plugins/plugin-list.md
index 4a540b5..b34cb6d 100755
--- a/docs/en/setup/plugins/plugin-list.md
+++ b/docs/en/setup/plugins/plugin-list.md
@@ -11,6 +11,8 @@
- Forwarder
- [envoy-als-v2-grpc-forwarder](./forwarder_envoy-als-v2-grpc-forwarder.md)
- [envoy-als-v3-grpc-forwarder](./forwarder_envoy-als-v3-grpc-forwarder.md)
+ - [envoy-metrics-v2-grpc-forwarder](./forwarder_envoy-metrics-v2-grpc-forwarder.md)
+ - [envoy-metrics-v3-grpc-forwarder](./forwarder_envoy-metrics-v3-grpc-forwarder.md)
- [nativecds-grpc-forwarder](./forwarder_nativecds-grpc-forwarder.md)
- [nativeevent-grpc-forwarder](./forwarder_nativeevent-grpc-forwarder.md)
- [nativejvm-grpc-forwarder](./forwarder_nativejvm-grpc-forwarder.md)
@@ -28,6 +30,8 @@
- Receiver
- [grpc-envoy-als-v2-receiver](./receiver_grpc-envoy-als-v2-receiver.md)
- [grpc-envoy-als-v3-receiver](./receiver_grpc-envoy-als-v3-receiver.md)
+ - [grpc-envoy-metrics-v2-receiver](./receiver_grpc-envoy-metrics-v2-receiver.md)
+ - [grpc-envoy-metrics-v3-receiver](./receiver_grpc-envoy-metrics-v3-receiver.md)
- [grpc-nativecds-receiver](./receiver_grpc-nativecds-receiver.md)
- [grpc-nativeevent-receiver](./receiver_grpc-nativeevent-receiver.md)
- [grpc-nativejvm-receiver](./receiver_grpc-nativejvm-receiver.md)
diff --git a/docs/en/setup/plugins/receiver_grpc-envoy-metrics-v2-receiver.md b/docs/en/setup/plugins/receiver_grpc-envoy-metrics-v2-receiver.md
new file mode 100755
index 0000000..132c339
--- /dev/null
+++ b/docs/en/setup/plugins/receiver_grpc-envoy-metrics-v2-receiver.md
@@ -0,0 +1,18 @@
+# Receiver/grpc-envoy-metrics-v2-receiver
+## Description
+This is a receiver for Envoy Metrics format, which is defined at https://github.com/envoyproxy/envoy/blob/v1.17.4/api/envoy/service/metrics/v2/metrics_service.proto.
+## Support Forwarders
+ - [envoy-metrics-v2-grpc-forwarder](forwarder_envoy-metrics-v2-grpc-forwarder.md)
+## DefaultConfig
+```yaml
+# The time interval between two flush operations. And the time unit is millisecond.
+flush_time: 1000
+# The max cache count when receive the message
+limit_count: 500
+```
+## Configuration
+|Name|Type|Description|
+|----|----|-----------|
+| flush_time | int | The time interval between two flush operations. And the time unit is millisecond. |
+| limit_count | int | The max cache count when receive the message |
+
diff --git a/docs/en/setup/plugins/receiver_grpc-envoy-metrics-v3-receiver.md b/docs/en/setup/plugins/receiver_grpc-envoy-metrics-v3-receiver.md
new file mode 100755
index 0000000..a327752
--- /dev/null
+++ b/docs/en/setup/plugins/receiver_grpc-envoy-metrics-v3-receiver.md
@@ -0,0 +1,18 @@
+# Receiver/grpc-envoy-metrics-v3-receiver
+## Description
+This is a receiver for Envoy Metrics format, which is defined at https://github.com/envoyproxy/envoy/blob/5f7d6efb5786ee3de31b1fb37c78fa281718b704/api/envoy/service/metrics/v3/metrics_service.proto.
+## Support Forwarders
+ - [envoy-metrics-v3-grpc-forwarder](forwarder_envoy-metrics-v3-grpc-forwarder.md)
+## DefaultConfig
+```yaml
+# The time interval between two flush operations. And the time unit is millisecond.
+flush_time: 1000
+# The max cache count when receive the message
+limit_count: 500
+```
+## Configuration
+|Name|Type|Description|
+|----|----|-----------|
+| flush_time | int | The time interval between two flush operations. And the time unit is millisecond. |
+| limit_count | int | The max cache count when receive the message |
+
diff --git a/docs/menu.yml b/docs/menu.yml
index f346f70..8d46699 100644
--- a/docs/menu.yml
+++ b/docs/menu.yml
@@ -79,6 +79,10 @@ catalog:
path: /en/setup/plugins/forwarder_envoy-als-v2-grpc-forwarder
- name: envoy als v3 grpc forwarder
path: /en/setup/plugins/forwarder_envoy-als-v3-grpc-forwarder
+ - name: envoy metrics v2 grpc forwarder
+ path: /en/setup/plugins/forwarder_envoy-metrics-v2-grpc-forwarder
+ - name: envoy metrics v3 grpc forwarder
+ path: /en/setup/plugins/forwarder_envoy-metrics-v3-grpc-forwarder
- name: nativecds grpc forwarder
path: /en/setup/plugins/forwarder_nativecds-grpc-forwarder
- name: nativeevent grpc forwarder
@@ -111,6 +115,10 @@ catalog:
path: /en/setup/plugins/receiver_grpc-envoy-als-v2-receiver
- name: grpc envoy als v3 receiver
path: /en/setup/plugins/receiver_grpc-envoy-als-v3-receiver
+ - name: grpc envoy metrics v2 receiver
+ path: /en/setup/plugins/receiver_grpc-envoy-metrics-v2-receiver
+ - name: grpc envoy metrics v3 receiver
+ path: /en/setup/plugins/receiver_grpc-envoy-metrics-v3-receiver
- name: grpc nativecds receiver
path: /en/setup/plugins/receiver_grpc-nativecds-receiver
- name: grpc nativeevent receiver
diff --git a/plugins/forwarder/forwarder_repository.go b/plugins/forwarder/forwarder_repository.go
index 19fbb32..3e3a9b9 100644
--- a/plugins/forwarder/forwarder_repository.go
+++ b/plugins/forwarder/forwarder_repository.go
@@ -22,6 +22,8 @@ import (
"github.com/apache/skywalking-satellite/plugins/forwarder/grpc/envoyalsv2"
"github.com/apache/skywalking-satellite/plugins/forwarder/grpc/envoyalsv3"
+ "github.com/apache/skywalking-satellite/plugins/forwarder/grpc/envoymetricsv2"
+ "github.com/apache/skywalking-satellite/plugins/forwarder/grpc/envoymetricsv3"
grpc_nativecds "github.com/apache/skywalking-satellite/plugins/forwarder/grpc/nativecds"
grpc_nativeevent "github.com/apache/skywalking-satellite/plugins/forwarder/grpc/nativeevent"
grpc_nativejvm "github.com/apache/skywalking-satellite/plugins/forwarder/grpc/nativejvm"
@@ -52,6 +54,8 @@ func RegisterForwarderPlugins() {
new(grpc_nativejvm.Forwarder),
new(envoyalsv2.Forwarder),
new(envoyalsv3.Forwarder),
+ new(envoymetricsv2.Forwarder),
+ new(envoymetricsv3.Forwarder),
}
for _, forwarder := range forwarders {
plugin.RegisterPlugin(forwarder)
diff --git a/plugins/forwarder/grpc/envoymetricsv2/forwarder.go b/plugins/forwarder/grpc/envoymetricsv2/forwarder.go
new file mode 100644
index 0000000..85b63aa
--- /dev/null
+++ b/plugins/forwarder/grpc/envoymetricsv2/forwarder.go
@@ -0,0 +1,106 @@
+// 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 envoymetricsv2
+
+import (
+ "context"
+ "fmt"
+ "io"
+ "reflect"
+
+ v2 "skywalking.apache.org/repo/goapi/proto/envoy/service/metrics/v2"
+ v1 "skywalking.apache.org/repo/goapi/satellite/data/v1"
+
+ "google.golang.org/grpc"
+
+ "github.com/apache/skywalking-satellite/internal/pkg/config"
+ "github.com/apache/skywalking-satellite/internal/pkg/log"
+ "github.com/apache/skywalking-satellite/internal/satellite/event"
+)
+
+const Name = "envoy-metrics-v2-grpc-forwarder"
+
+type Forwarder struct {
+ config.CommonFields
+ metricsClient v2.MetricsServiceClient
+}
+
+func (f *Forwarder) Name() string {
+ return Name
+}
+
+func (f *Forwarder) Description() string {
+ return "This is a synchronization Metrics v2 grpc forwarder with the Envoy metrics protocol."
+}
+
+func (f *Forwarder) DefaultConfig() string {
+ return ``
+}
+
+func (f *Forwarder) Prepare(connection interface{}) error {
+ client, ok := connection.(*grpc.ClientConn)
+ if !ok {
+ return fmt.Errorf("the %s only accepts a grpc client, but received a %s",
+ f.Name(), reflect.TypeOf(connection).String())
+ }
+ f.metricsClient = v2.NewMetricsServiceClient(client)
+ return nil
+}
+
+func (f *Forwarder) Forward(batch event.BatchEvents) error {
+ for _, e := range batch {
+ data, ok := e.GetData().(*v1.SniffData_EnvoyMetricsV2List)
+ if !ok {
+ continue
+ }
+ stream, err := f.metricsClient.StreamMetrics(context.Background())
+ if err != nil {
+ log.Logger.Errorf("open grpc stream error %v", err)
+ return err
+ }
+ for _, message := range data.EnvoyMetricsV2List.Messages {
+ err := stream.Send(message)
+ if err != nil {
+ log.Logger.Errorf("%s send envoy metrics v2 data error: %v", f.Name(), err)
+ f.closeStream(stream)
+ return err
+ }
+ }
+ f.closeStream(stream)
+ }
+ return nil
+}
+
+func (f *Forwarder) closeStream(stream v2.MetricsService_StreamMetricsClient) {
+ _, err := stream.CloseAndRecv()
+ if err != nil && err != io.EOF {
+ log.Logger.Warnf("%s close stream error: %v", f.Name(), err)
+ }
+}
+
+func (f *Forwarder) ForwardType() v1.SniffType {
+ return v1.SniffType_EnvoyMetricsV2Type
+}
+
+func (f *Forwarder) SyncForward(_ *v1.SniffData) (*v1.SniffData, error) {
+ return nil, fmt.Errorf("unsupport sync forward")
+}
+
+func (f *Forwarder) SupportedSyncInvoke() bool {
+ return false
+}
diff --git a/plugins/forwarder/grpc/envoymetricsv3/forwarder.go b/plugins/forwarder/grpc/envoymetricsv3/forwarder.go
new file mode 100644
index 0000000..a267a04
--- /dev/null
+++ b/plugins/forwarder/grpc/envoymetricsv3/forwarder.go
@@ -0,0 +1,106 @@
+// 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 envoymetricsv3
+
+import (
+ "context"
+ "fmt"
+ "io"
+ "reflect"
+
+ v3 "skywalking.apache.org/repo/goapi/proto/envoy/service/metrics/v3"
+ v1 "skywalking.apache.org/repo/goapi/satellite/data/v1"
+
+ "google.golang.org/grpc"
+
+ "github.com/apache/skywalking-satellite/internal/pkg/config"
+ "github.com/apache/skywalking-satellite/internal/pkg/log"
+ "github.com/apache/skywalking-satellite/internal/satellite/event"
+)
+
+const Name = "envoy-metrics-v3-grpc-forwarder"
+
+type Forwarder struct {
+ config.CommonFields
+ metricsClient v3.MetricsServiceClient
+}
+
+func (f *Forwarder) Name() string {
+ return Name
+}
+
+func (f *Forwarder) Description() string {
+ return "This is a synchronization Metrics v3 grpc forwarder with the Envoy metrics protocol."
+}
+
+func (f *Forwarder) DefaultConfig() string {
+ return ``
+}
+
+func (f *Forwarder) Prepare(connection interface{}) error {
+ client, ok := connection.(*grpc.ClientConn)
+ if !ok {
+ return fmt.Errorf("the %s only accepts a grpc client, but received a %s",
+ f.Name(), reflect.TypeOf(connection).String())
+ }
+ f.metricsClient = v3.NewMetricsServiceClient(client)
+ return nil
+}
+
+func (f *Forwarder) Forward(batch event.BatchEvents) error {
+ for _, e := range batch {
+ data, ok := e.GetData().(*v1.SniffData_EnvoyMetricsV3List)
+ if !ok {
+ continue
+ }
+ stream, err := f.metricsClient.StreamMetrics(context.Background())
+ if err != nil {
+ log.Logger.Errorf("open grpc stream error %v", err)
+ return err
+ }
+ for _, message := range data.EnvoyMetricsV3List.Messages {
+ err := stream.Send(message)
+ if err != nil {
+ log.Logger.Errorf("%s send envoy metrics v3 data error: %v", f.Name(), err)
+ f.closeStream(stream)
+ return err
+ }
+ }
+ f.closeStream(stream)
+ }
+ return nil
+}
+
+func (f *Forwarder) closeStream(stream v3.MetricsService_StreamMetricsClient) {
+ _, err := stream.CloseAndRecv()
+ if err != nil && err != io.EOF {
+ log.Logger.Warnf("%s close stream error: %v", f.Name(), err)
+ }
+}
+
+func (f *Forwarder) ForwardType() v1.SniffType {
+ return v1.SniffType_EnvoyMetricsV3Type
+}
+
+func (f *Forwarder) SyncForward(_ *v1.SniffData) (*v1.SniffData, error) {
+ return nil, fmt.Errorf("unsupport sync forward")
+}
+
+func (f *Forwarder) SupportedSyncInvoke() bool {
+ return false
+}
diff --git a/plugins/receiver/grpc/envoymetricsv2/als_service.go b/plugins/receiver/grpc/envoymetricsv2/als_service.go
new file mode 100644
index 0000000..d8d7dcc
--- /dev/null
+++ b/plugins/receiver/grpc/envoymetricsv2/als_service.go
@@ -0,0 +1,92 @@
+// 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 envoymetricsv2
+
+import (
+ "context"
+ "io"
+ "time"
+
+ "github.com/apache/skywalking-satellite/internal/satellite/module/buffer"
+
+ v2 "skywalking.apache.org/repo/goapi/proto/envoy/service/metrics/v2"
+ v1 "skywalking.apache.org/repo/goapi/satellite/data/v1"
+)
+
+const eventName = "grpc-envoy-metrics-v2-event"
+
+type MetricsService struct {
+ receiveChannel chan *v1.SniffData
+ limiterConfig buffer.LimiterConfig
+ v2.UnimplementedMetricsServiceServer
+}
+
+func (m *MetricsService) StreamMetrics(stream v2.MetricsService_StreamMetricsServer) error {
+ messages := make(chan *v2.StreamMetricsMessage, m.limiterConfig.LimitCount*2)
+ limiter := buffer.NewLimiter(m.limiterConfig, func() int {
+ return len(messages)
+ })
+
+ var identity *v2.StreamMetricsMessage_Identifier
+
+ defer limiter.Stop()
+ limiter.Start(context.Background(), func() {
+ count := len(messages)
+ if count == 0 {
+ return
+ }
+ metricsMessages := make([]*v2.StreamMetricsMessage, 0)
+ for i := 0; i < count; i++ {
+ metricsMessages = append(metricsMessages, <-messages)
+ }
+ metricsMessages[0].Identifier = identity
+
+ d := &v1.SniffData{
+ Name: eventName,
+ Timestamp: time.Now().UnixNano() / 1e6,
+ Meta: nil,
+ Type: v1.SniffType_EnvoyMetricsV2Type,
+ Remote: true,
+ Data: &v1.SniffData_EnvoyMetricsV2List{
+ EnvoyMetricsV2List: &v1.EnvoyMetricsV2List{
+ Messages: metricsMessages,
+ },
+ },
+ }
+ m.receiveChannel <- d
+ })
+
+ var err1 error
+ for {
+ item, err := stream.Recv()
+ if err != nil {
+ err1 = err
+ break
+ }
+ if item.Identifier != nil {
+ identity = item.Identifier
+ }
+ messages <- item
+ limiter.Check()
+ }
+
+ if err1 != io.EOF {
+ return err1
+ }
+ return stream.SendAndClose(&v2.StreamMetricsResponse{})
+}
diff --git a/plugins/receiver/grpc/envoymetricsv2/receiver.go b/plugins/receiver/grpc/envoymetricsv2/receiver.go
new file mode 100644
index 0000000..82b5560
--- /dev/null
+++ b/plugins/receiver/grpc/envoymetricsv2/receiver.go
@@ -0,0 +1,77 @@
+// 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 envoymetricsv2
+
+import (
+ "github.com/apache/skywalking-satellite/internal/pkg/config"
+ module "github.com/apache/skywalking-satellite/internal/satellite/module/api"
+ "github.com/apache/skywalking-satellite/internal/satellite/module/buffer"
+ forwarder "github.com/apache/skywalking-satellite/plugins/forwarder/api"
+ "github.com/apache/skywalking-satellite/plugins/forwarder/grpc/envoymetricsv2"
+ "github.com/apache/skywalking-satellite/plugins/receiver/grpc"
+
+ v2 "skywalking.apache.org/repo/goapi/proto/envoy/service/metrics/v2"
+ v1 "skywalking.apache.org/repo/goapi/satellite/data/v1"
+)
+
+const Name = "grpc-envoy-metrics-v2-receiver"
+
+type Receiver struct {
+ config.CommonFields
+ grpc.CommonGRPCReceiverFields
+ service *MetricsService
+
+ LimitConfig buffer.LimiterConfig `mapstructure:",squash"`
+}
+
+func (r *Receiver) Name() string {
+ return Name
+}
+
+func (r *Receiver) Description() string {
+ return "This is a receiver for Envoy Metrics format, " +
+ "which is defined at https://github.com/envoyproxy/envoy/blob/v1.17.4/api/envoy/service/metrics/v2/metrics_service.proto."
+}
+
+func (r *Receiver) DefaultConfig() string {
+ return `
+# The time interval between two flush operations. And the time unit is millisecond.
+flush_time: 1000
+# The max cache count when receive the message
+limit_count: 500
+`
+}
+
+func (r *Receiver) RegisterHandler(server interface{}) {
+ r.CommonGRPCReceiverFields = *grpc.InitCommonGRPCReceiverFields(server)
+ r.service = &MetricsService{receiveChannel: r.OutputChannel, limiterConfig: r.LimitConfig}
+ v2.RegisterMetricsServiceServer(r.Server, r.service)
+}
+
+func (r *Receiver) RegisterSyncInvoker(_ module.SyncInvoker) {
+}
+
+func (r *Receiver) Channel() <-chan *v1.SniffData {
+ return r.OutputChannel
+}
+
+func (r *Receiver) SupportForwarders() []forwarder.Forwarder {
+ return []forwarder.Forwarder{
+ new(envoymetricsv2.Forwarder),
+ }
+}
diff --git a/plugins/receiver/grpc/envoymetricsv2/receiver_test.go b/plugins/receiver/grpc/envoymetricsv2/receiver_test.go
new file mode 100644
index 0000000..d8274aa
--- /dev/null
+++ b/plugins/receiver/grpc/envoymetricsv2/receiver_test.go
@@ -0,0 +1,67 @@
+// 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 envoymetricsv2
+
+import (
+ "context"
+ "strconv"
+ "testing"
+
+ "skywalking.apache.org/repo/goapi/proto/envoy/api/v2/core"
+
+ v2 "skywalking.apache.org/repo/goapi/proto/envoy/service/metrics/v2"
+ v1 "skywalking.apache.org/repo/goapi/satellite/data/v1"
+
+ "google.golang.org/grpc"
+
+ _ "github.com/apache/skywalking-satellite/internal/satellite/test"
+ receiver_grpc "github.com/apache/skywalking-satellite/plugins/receiver/grpc"
+)
+
+func TestReceiver_RegisterHandler(t *testing.T) {
+ recConf := make(map[string]string, 2)
+ recConf["limit_count"] = "1"
+ recConf["flush_time"] = "1000"
+ receiver_grpc.TestReceiverWithConfig(new(Receiver), recConf, func(t *testing.T, sequence int, conn *grpc.ClientConn, ctx context.Context) string {
+ client := v2.NewMetricsServiceClient(conn)
+ data := initData(sequence)
+ collect, err := client.StreamMetrics(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.GetEnvoyMetricsV2List().Messages[0].String()
+ }, t)
+}
+
+func initData(sequence int) *v2.StreamMetricsMessage {
+ return &v2.StreamMetricsMessage{
+ Identifier: &v2.StreamMetricsMessage_Identifier{
+ Node: &core.Node{
+ Id: "test" + strconv.Itoa(sequence),
+ },
+ },
+ }
+}
diff --git a/plugins/receiver/grpc/envoymetricsv3/als_service.go b/plugins/receiver/grpc/envoymetricsv3/als_service.go
new file mode 100644
index 0000000..b3cc830
--- /dev/null
+++ b/plugins/receiver/grpc/envoymetricsv3/als_service.go
@@ -0,0 +1,92 @@
+// 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 envoymetricsv3
+
+import (
+ "context"
+ "io"
+ "time"
+
+ "github.com/apache/skywalking-satellite/internal/satellite/module/buffer"
+
+ v3 "skywalking.apache.org/repo/goapi/proto/envoy/service/metrics/v3"
+ v1 "skywalking.apache.org/repo/goapi/satellite/data/v1"
+)
+
+const eventName = "grpc-envoy-metrics-v3-event"
+
+type MetricsService struct {
+ receiveChannel chan *v1.SniffData
+ limiterConfig buffer.LimiterConfig
+ v3.UnimplementedMetricsServiceServer
+}
+
+func (m *MetricsService) StreamMetrics(stream v3.MetricsService_StreamMetricsServer) error {
+ messages := make(chan *v3.StreamMetricsMessage, m.limiterConfig.LimitCount*2)
+ limiter := buffer.NewLimiter(m.limiterConfig, func() int {
+ return len(messages)
+ })
+
+ var identity *v3.StreamMetricsMessage_Identifier
+
+ defer limiter.Stop()
+ limiter.Start(context.Background(), func() {
+ count := len(messages)
+ if count == 0 {
+ return
+ }
+ metricsMessages := make([]*v3.StreamMetricsMessage, 0)
+ for i := 0; i < count; i++ {
+ metricsMessages = append(metricsMessages, <-messages)
+ }
+ metricsMessages[0].Identifier = identity
+
+ d := &v1.SniffData{
+ Name: eventName,
+ Timestamp: time.Now().UnixNano() / 1e6,
+ Meta: nil,
+ Type: v1.SniffType_EnvoyMetricsV3Type,
+ Remote: true,
+ Data: &v1.SniffData_EnvoyMetricsV3List{
+ EnvoyMetricsV3List: &v1.EnvoyMetricsV3List{
+ Messages: metricsMessages,
+ },
+ },
+ }
+ m.receiveChannel <- d
+ })
+
+ var err1 error
+ for {
+ item, err := stream.Recv()
+ if err != nil {
+ err1 = err
+ break
+ }
+ if item.Identifier != nil {
+ identity = item.Identifier
+ }
+ messages <- item
+ limiter.Check()
+ }
+
+ if err1 != io.EOF {
+ return err1
+ }
+ return stream.SendAndClose(&v3.StreamMetricsResponse{})
+}
diff --git a/plugins/receiver/grpc/envoymetricsv3/receiver.go b/plugins/receiver/grpc/envoymetricsv3/receiver.go
new file mode 100644
index 0000000..b3da8b6
--- /dev/null
+++ b/plugins/receiver/grpc/envoymetricsv3/receiver.go
@@ -0,0 +1,78 @@
+// 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 envoymetricsv3
+
+import (
+ "github.com/apache/skywalking-satellite/internal/pkg/config"
+ module "github.com/apache/skywalking-satellite/internal/satellite/module/api"
+ "github.com/apache/skywalking-satellite/internal/satellite/module/buffer"
+ forwarder "github.com/apache/skywalking-satellite/plugins/forwarder/api"
+ "github.com/apache/skywalking-satellite/plugins/forwarder/grpc/envoymetricsv3"
+ "github.com/apache/skywalking-satellite/plugins/receiver/grpc"
+
+ v3 "skywalking.apache.org/repo/goapi/proto/envoy/service/metrics/v3"
+ v1 "skywalking.apache.org/repo/goapi/satellite/data/v1"
+)
+
+const Name = "grpc-envoy-metrics-v3-receiver"
+
+type Receiver struct {
+ config.CommonFields
+ grpc.CommonGRPCReceiverFields
+ service *MetricsService
+
+ LimitConfig buffer.LimiterConfig `mapstructure:",squash"`
+}
+
+func (r *Receiver) Name() string {
+ return Name
+}
+
+func (r *Receiver) Description() string {
+ return "This is a receiver for Envoy Metrics format, " +
+ "which is defined at https://github.com/envoyproxy/envoy/blob/" +
+ "5f7d6efb5786ee3de31b1fb37c78fa281718b704/api/envoy/service/metrics/v3/metrics_service.proto."
+}
+
+func (r *Receiver) DefaultConfig() string {
+ return `
+# The time interval between two flush operations. And the time unit is millisecond.
+flush_time: 1000
+# The max cache count when receive the message
+limit_count: 500
+`
+}
+
+func (r *Receiver) RegisterHandler(server interface{}) {
+ r.CommonGRPCReceiverFields = *grpc.InitCommonGRPCReceiverFields(server)
+ r.service = &MetricsService{receiveChannel: r.OutputChannel, limiterConfig: r.LimitConfig}
+ v3.RegisterMetricsServiceServer(r.Server, r.service)
+}
+
+func (r *Receiver) RegisterSyncInvoker(_ module.SyncInvoker) {
+}
+
+func (r *Receiver) Channel() <-chan *v1.SniffData {
+ return r.OutputChannel
+}
+
+func (r *Receiver) SupportForwarders() []forwarder.Forwarder {
+ return []forwarder.Forwarder{
+ new(envoymetricsv3.Forwarder),
+ }
+}
diff --git a/plugins/receiver/grpc/envoymetricsv3/receiver_test.go b/plugins/receiver/grpc/envoymetricsv3/receiver_test.go
new file mode 100644
index 0000000..2cbc092
--- /dev/null
+++ b/plugins/receiver/grpc/envoymetricsv3/receiver_test.go
@@ -0,0 +1,66 @@
+// 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 envoymetricsv3
+
+import (
+ "context"
+ "strconv"
+ "testing"
+
+ corev3 "skywalking.apache.org/repo/goapi/proto/envoy/config/core/v3"
+ v3 "skywalking.apache.org/repo/goapi/proto/envoy/service/metrics/v3"
+ v1 "skywalking.apache.org/repo/goapi/satellite/data/v1"
+
+ "google.golang.org/grpc"
+
+ _ "github.com/apache/skywalking-satellite/internal/satellite/test"
+ receiver_grpc "github.com/apache/skywalking-satellite/plugins/receiver/grpc"
+)
+
+func TestReceiver_RegisterHandler(t *testing.T) {
+ recConf := make(map[string]string, 2)
+ recConf["limit_count"] = "1"
+ recConf["flush_time"] = "1000"
+ receiver_grpc.TestReceiverWithConfig(new(Receiver), recConf, func(t *testing.T, sequence int, conn *grpc.ClientConn, ctx context.Context) string {
+ client := v3.NewMetricsServiceClient(conn)
+ data := initData(sequence)
+ collect, err := client.StreamMetrics(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.GetEnvoyMetricsV3List().Messages[0].String()
+ }, t)
+}
+
+func initData(sequence int) *v3.StreamMetricsMessage {
+ return &v3.StreamMetricsMessage{
+ Identifier: &v3.StreamMetricsMessage_Identifier{
+ Node: &corev3.Node{
+ Id: "test" + strconv.Itoa(sequence),
+ },
+ },
+ }
+}
diff --git a/plugins/receiver/receiver_repository.go b/plugins/receiver/receiver_repository.go
index 0b82989..9841ed2 100644
--- a/plugins/receiver/receiver_repository.go
+++ b/plugins/receiver/receiver_repository.go
@@ -24,6 +24,8 @@ import (
"github.com/apache/skywalking-satellite/plugins/receiver/api"
"github.com/apache/skywalking-satellite/plugins/receiver/grpc/envoyalsv2"
"github.com/apache/skywalking-satellite/plugins/receiver/grpc/envoyalsv3"
+ "github.com/apache/skywalking-satellite/plugins/receiver/grpc/envoymetricsv2"
+ "github.com/apache/skywalking-satellite/plugins/receiver/grpc/envoymetricsv3"
grpcnativecds "github.com/apache/skywalking-satellite/plugins/receiver/grpc/nativecds"
grpcnativeevent "github.com/apache/skywalking-satellite/plugins/receiver/grpc/nativeevent"
grpcnativejvm "github.com/apache/skywalking-satellite/plugins/receiver/grpc/nativejvm"
@@ -51,6 +53,8 @@ func RegisterReceiverPlugins() {
new(grpcnativemeter.Receiver),
new(envoyalsv2.Receiver),
new(envoyalsv3.Receiver),
+ new(envoymetricsv2.Receiver),
+ new(envoymetricsv3.Receiver),
}
for _, receiver := range receivers {
plugin.RegisterPlugin(receiver)
diff --git a/test/e2e/case/envoy/e2e.yaml b/test/e2e/case/envoy/e2e.yaml
index a2815ef..90c395a 100644
--- a/test/e2e/case/envoy/e2e.yaml
+++ b/test/e2e/case/envoy/e2e.yaml
@@ -28,7 +28,7 @@ setup:
cleanup:
# always never success failure
- on: no
+ on: always
trigger:
action: http
@@ -53,4 +53,7 @@ verify:
expected: expected/service-instance.yml
# als: logs list
- query: swctl --display yaml --base-url=http://${oap_host}:${oap_12800}/graphql logs list --service-name test-app-v1 |yq e 'del(.. | select(has("content")).content)' -
- expected: expected/logs-list.yml
\ No newline at end of file
+ expected: expected/logs-list.yml
+ # metrics: envoy metrics
+ - query: swctl --display yaml --base-url=http://${oap_host}:${oap_12800}/graphql metrics linear --name envoy_worker_threads --service-name istio-dp::test-app-v1 --instance-name service-instance-name-v1|yq e 'to_entries' -
+ expected: expected/metrics-has-value.yml
\ No newline at end of file
diff --git a/test/e2e/case/envoy/envoy-v1.16.2.yaml b/test/e2e/case/envoy/envoy-v1.16.2.yaml
index fbd87fc..9bf7c17 100644
--- a/test/e2e/case/envoy/envoy-v1.16.2.yaml
+++ b/test/e2e/case/envoy/envoy-v1.16.2.yaml
@@ -16,6 +16,13 @@
# under the License.
#
+stats_sinks:
+ - name: envoy.metrics_service
+ config:
+ grpc_service:
+ envoy_grpc:
+ cluster_name: skywalking-satellite
+
node:
id: ingress-v1
cluster: card-api-cluster-v1
diff --git a/test/e2e/case/envoy/envoy-v1.19.0.yaml b/test/e2e/case/envoy/envoy-v1.19.0.yaml
index 4d16c5f..74ff15b 100644
--- a/test/e2e/case/envoy/envoy-v1.19.0.yaml
+++ b/test/e2e/case/envoy/envoy-v1.19.0.yaml
@@ -16,6 +16,15 @@
# under the License.
#
+stats_sinks:
+ - name: envoy.stat_sinks.metrics_service
+ typed_config:
+ "@type": type.googleapis.com/envoy.config.metrics.v3.MetricsServiceConfig
+ transport_api_version: V3
+ grpc_service:
+ envoy_grpc:
+ cluster_name: skywalking-satellite
+
node:
id: ingress-v1
cluster: card-api-cluster-v1
diff --git a/test/e2e/case/envoy/expected/metrics-has-value.yml b/test/e2e/case/envoy/expected/metrics-has-value.yml
new file mode 100644
index 0000000..d9c4985
--- /dev/null
+++ b/test/e2e/case/envoy/expected/metrics-has-value.yml
@@ -0,0 +1,19 @@
+# Licensed to the 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.
+# The 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.
+
+{{- contains . }}
+- key: {{ notEmpty .key }}
+ value: {{ ge .value 1 }}
+{{- end }}
\ No newline at end of file