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/11/09 15:24:38 UTC
[skywalking-satellite] branch main updated: Support partition queue
(#82)
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 b26c828 Support partition queue (#82)
b26c828 is described below
commit b26c828d485998b9eef355e4ba0a5634d0fdccd1
Author: mrproliu <74...@qq.com>
AuthorDate: Tue Nov 9 23:24:32 2021 +0800
Support partition queue (#82)
---
CHANGES.md | 1 +
configs/satellite_config.yaml | 36 +++++
docs/en/concepts-and-designs/README.md | 4 +-
docs/en/setup/README.md | 5 +-
.../README.md | 2 +-
.../satellite_config.yaml | 0
.../examples/grpc-load-balance-client/README.md | 2 +-
.../setup/examples/high-throughput-queue/README.md | 19 +++
.../setup/examples/transmit-log-to-kafka/README.md | 2 +-
docs/en/setup/plugins/plugin-list.md | 74 ++++-----
docs/en/setup/plugins/queue_memory-queue.md | 4 +
docs/en/setup/plugins/queue_mmap-queue.md | 4 +
docs/en/setup/plugins/queue_none-queue.md | 6 +-
docs/menu.yml | 110 +++++++-------
internal/pkg/plugin/definition.go | 2 +
.../democategory/demodir1/demodir2/plugin_test.go | 4 +
internal/pkg/plugin/registry.go | 6 +-
internal/satellite/config/loader_test.go | 1 +
internal/satellite/event/event.go | 14 +-
internal/satellite/module/buffer/buffer.go | 17 +--
internal/satellite/module/buffer/buffer_test.go | 10 +-
internal/satellite/module/gatherer/api/gatherer.go | 6 +-
internal/satellite/module/gatherer/create.go | 8 +-
.../satellite/module/gatherer/fetcher_gatherer.go | 32 +++-
.../satellite/module/gatherer/receiver_gatherer.go | 33 ++--
internal/satellite/module/processor/processor.go | 14 +-
internal/satellite/module/sender/api/sender.go | 2 +-
internal/satellite/module/sender/create.go | 5 -
internal/satellite/module/sender/sender.go | 95 ++++++++----
internal/satellite/telemetry/gauge.go | 58 +++++++
internal/satellite/telemetry/timer.go | 76 +++++++++
internal/satellite/tools/generate_plugin_doc.go | 71 +++++++--
plugins/client/grpc/client.go | 9 +-
plugins/client/kafka/client.go | 9 +-
plugins/fallbacker/none/none_fallbacker.go | 9 +-
plugins/fallbacker/timer/timer_fallbacker.go | 9 +-
plugins/fetcher/prometheus/fetcher.go | 5 +
plugins/forwarder/grpc/envoyalsv2/forwarder.go | 9 +-
plugins/forwarder/grpc/envoyalsv3/forwarder.go | 67 +++++++-
plugins/forwarder/grpc/envoymetricsv2/forwarder.go | 9 +-
plugins/forwarder/grpc/envoymetricsv3/forwarder.go | 9 +-
plugins/forwarder/grpc/nativecds/forwarder.go | 9 +-
.../forwarder/grpc/nativeevent/sync_forwarder.go | 9 +-
plugins/forwarder/grpc/nativejvm/sync_forwarder.go | 9 +-
plugins/forwarder/grpc/nativelog/forwarder.go | 9 +-
.../forwarder/grpc/nativemanagement/forwarder.go | 9 +-
plugins/forwarder/grpc/nativemeter/forwarder.go | 9 +-
plugins/forwarder/grpc/nativeprofile/forwarder.go | 9 +-
plugins/forwarder/grpc/nativetracing/forwarder.go | 9 +-
plugins/forwarder/kafka/nativelog/forwarder.go | 9 +-
plugins/queue/api/queue.go | 13 +-
plugins/queue/memory/queue.go | 27 +++-
plugins/queue/mmap/queue.go | 28 +++-
plugins/queue/mmap/queue_operation.go | 9 +-
plugins/queue/mmap/segment/segment.go | 2 +-
plugins/queue/none/queue.go | 21 ++-
plugins/queue/partition/partitioned_queue.go | 169 +++++++++++++++++++++
plugins/queue/partition/partitioned_queue_test.go | 145 ++++++++++++++++++
plugins/receiver/grpc/envoyalsv2/receiver.go | 9 +-
plugins/receiver/grpc/envoyalsv3/als_service.go | 25 +++
plugins/receiver/grpc/envoyalsv3/receiver.go | 10 +-
plugins/receiver/grpc/envoymetricsv2/receiver.go | 9 +-
plugins/receiver/grpc/envoymetricsv3/receiver.go | 9 +-
plugins/receiver/grpc/nativecds/receiver.go | 9 +-
plugins/receiver/grpc/nativeevent/receiver.go | 9 +-
plugins/receiver/grpc/nativejvm/receiver.go | 9 +-
plugins/receiver/grpc/nativelog/receiver.go | 9 +-
plugins/receiver/grpc/nativemanagement/receiver.go | 9 +-
plugins/receiver/grpc/nativemeter/receiver.go | 9 +-
plugins/receiver/grpc/nativeprofile/receiver.go | 9 +-
plugins/receiver/grpc/nativetracing/receiver.go | 9 +-
plugins/receiver/http/nativcelog/receiver.go | 5 +
plugins/server/grpc/server.go | 9 +-
plugins/server/http/server.go | 9 +-
plugins/server/prometheus/prometheus.go | 9 +-
75 files changed, 1270 insertions(+), 249 deletions(-)
diff --git a/CHANGES.md b/CHANGES.md
index 9f1d6a7..61781b9 100644
--- a/CHANGES.md
+++ b/CHANGES.md
@@ -5,6 +5,7 @@ Release Notes.
0.4.0
------------------
#### Features
+* Support partition queue.
#### Bug Fixes
diff --git a/configs/satellite_config.yaml b/configs/satellite_config.yaml
index c7ec8e8..db9fcef 100644
--- a/configs/satellite_config.yaml
+++ b/configs/satellite_config.yaml
@@ -79,6 +79,8 @@ pipes:
plugin_name: "memory-queue"
# The maximum buffer event size.
event_buffer_size: ${SATELLITE_QUEUE_EVENT_BUFFER_SIZE:5000}
+ # The partition count of queue.
+ partition: ${SATELLITE_QUEUE_PARTITION:1}
processor:
filters:
sender:
@@ -103,6 +105,8 @@ pipes:
plugin_name: "memory-queue"
# The maximum buffer event size.
event_buffer_size: ${SATELLITE_QUEUE_EVENT_BUFFER_SIZE:5000}
+ # The partition count of queue.
+ partition: ${SATELLITE_QUEUE_PARTITION:1}
processor:
filters:
sender:
@@ -127,6 +131,8 @@ pipes:
plugin_name: "memory-queue"
# The maximum buffer event size.
event_buffer_size: ${SATELLITE_QUEUE_EVENT_BUFFER_SIZE:5000}
+ # The partition count of queue.
+ partition: ${SATELLITE_QUEUE_PARTITION:1}
processor:
filters:
sender:
@@ -151,6 +157,8 @@ pipes:
plugin_name: "memory-queue"
# The maximum buffer event size.
event_buffer_size: ${SATELLITE_QUEUE_EVENT_BUFFER_SIZE:5000}
+ # The partition count of queue.
+ partition: ${SATELLITE_QUEUE_PARTITION:1}
processor:
filters:
sender:
@@ -189,6 +197,10 @@ pipes:
plugin_name: "grpc-nativeevent-receiver"
queue:
plugin_name: "memory-queue"
+ # The maximum buffer event size.
+ event_buffer_size: ${SATELLITE_QUEUE_EVENT_BUFFER_SIZE:5000}
+ # The partition count of queue.
+ partition: ${SATELLITE_QUEUE_PARTITION:1}
processor:
filters:
sender:
@@ -211,6 +223,10 @@ pipes:
plugin_name: "grpc-nativejvm-receiver"
queue:
plugin_name: "memory-queue"
+ # The maximum buffer event size.
+ event_buffer_size: ${SATELLITE_QUEUE_EVENT_BUFFER_SIZE:5000}
+ # The partition count of queue.
+ partition: ${SATELLITE_QUEUE_PARTITION:1}
processor:
filters:
sender:
@@ -233,6 +249,10 @@ pipes:
plugin_name: "grpc-nativemeter-receiver"
queue:
plugin_name: "memory-queue"
+ # The maximum buffer event size.
+ event_buffer_size: ${SATELLITE_QUEUE_EVENT_BUFFER_SIZE:5000}
+ # The partition count of queue.
+ partition: ${SATELLITE_QUEUE_PARTITION:1}
processor:
filters:
sender:
@@ -259,6 +279,10 @@ pipes:
limit_count: ${SATELLITE_ENVOY_ALS_V2_PIPE_RECEIVER_LIMIT_COUNT:500}
queue:
plugin_name: "memory-queue"
+ # The maximum buffer event size.
+ event_buffer_size: ${SATELLITE_QUEUE_EVENT_BUFFER_SIZE:5000}
+ # The partition count of queue.
+ partition: ${SATELLITE_QUEUE_PARTITION:1}
processor:
filters:
sender:
@@ -285,6 +309,10 @@ pipes:
limit_count: ${SATELLITE_ENVOY_ALS_V3_PIPE_RECEIVER_LIMIT_COUNT:500}
queue:
plugin_name: "memory-queue"
+ # The maximum buffer event size.
+ event_buffer_size: ${SATELLITE_QUEUE_EVENT_BUFFER_SIZE:5000}
+ # The partition count of queue.
+ partition: ${SATELLITE_QUEUE_PARTITION:1}
processor:
filters:
sender:
@@ -311,6 +339,10 @@ pipes:
limit_count: ${SATELLITE_ENVOY_METRICS_V2_PIPE_RECEIVER_LIMIT_COUNT:500}
queue:
plugin_name: "memory-queue"
+ # The maximum buffer event size.
+ event_buffer_size: ${SATELLITE_QUEUE_EVENT_BUFFER_SIZE:5000}
+ # The partition count of queue.
+ partition: ${SATELLITE_QUEUE_PARTITION:1}
processor:
filters:
sender:
@@ -337,6 +369,10 @@ pipes:
limit_count: ${SATELLITE_ENVOY_METRICS_V3_PIPE_RECEIVER_LIMIT_COUNT:500}
queue:
plugin_name: "memory-queue"
+ # The maximum buffer event size.
+ event_buffer_size: ${SATELLITE_QUEUE_EVENT_BUFFER_SIZE:5000}
+ # The partition count of queue.
+ partition: ${SATELLITE_QUEUE_PARTITION:1}
processor:
filters:
sender:
diff --git a/docs/en/concepts-and-designs/README.md b/docs/en/concepts-and-designs/README.md
index 084d53c..1d6d748 100644
--- a/docs/en/concepts-and-designs/README.md
+++ b/docs/en/concepts-and-designs/README.md
@@ -9,6 +9,6 @@ After you read the above documents, you should understand basic goals of the Sky
you are interested, then dive in.
- [Module Design](./module_design.md)
-- [Plugin mechanism](./plugin_mechanism.md)
+- [Plugin Mechanism](./plugin_mechanism.md)
- [Project Structure](./project_structue.md)
-- [The design of the memory mapped queue](./mmap-queue.md)
+- [Memory mapped Queue](./mmap-queue.md)
diff --git a/docs/en/setup/README.md b/docs/en/setup/README.md
index 1b0a23a..dc1aa72 100644
--- a/docs/en/setup/README.md
+++ b/docs/en/setup/README.md
@@ -19,9 +19,10 @@ bin/startup.sh
## Examples
You can quickly build your satellite according to the following examples:
-1. [Transmit protocol from agent](examples/transmit-protocol-from-agent/README.md)
+1. [Forward Agent Data to OAP Server](examples/forward-agent-data-to-oap-server/README.md)
2. [Transmit Log to Kafka](examples/transmit-log-to-kafka/README.md)
-3. [GRPC load balance client](examples/grpc-load-balance-client/README.md)
+3. [GRPC load balance Client](examples/grpc-load-balance-client/README.md)
+3. [High Throughput Queue](examples/high-throughput-queue/README.md)
## satellite_config.yaml
The core concept behind this setting file is, SkyWalking Satellite is based on pure modularization design. End user can switch or assemble the collector features by their own requirements.
diff --git a/docs/en/setup/examples/transmit-protocol-from-agent/README.md b/docs/en/setup/examples/forward-agent-data-to-oap-server/README.md
similarity index 92%
rename from docs/en/setup/examples/transmit-protocol-from-agent/README.md
rename to docs/en/setup/examples/forward-agent-data-to-oap-server/README.md
index 80f1a2e..5cce412 100644
--- a/docs/en/setup/examples/transmit-protocol-from-agent/README.md
+++ b/docs/en/setup/examples/forward-agent-data-to-oap-server/README.md
@@ -1,4 +1,4 @@
-# Transmit protocol from agent
+# Forward Agent Data to OAP Server
Using Satellite to receive the SkyWalking protocols from agent, and transport data to the SkyWalking backend or another Satellite.
diff --git a/docs/en/setup/examples/transmit-protocol-from-agent/satellite_config.yaml b/docs/en/setup/examples/forward-agent-data-to-oap-server/satellite_config.yaml
similarity index 100%
rename from docs/en/setup/examples/transmit-protocol-from-agent/satellite_config.yaml
rename to docs/en/setup/examples/forward-agent-data-to-oap-server/satellite_config.yaml
diff --git a/docs/en/setup/examples/grpc-load-balance-client/README.md b/docs/en/setup/examples/grpc-load-balance-client/README.md
index d792608..03ed2bd 100644
--- a/docs/en/setup/examples/grpc-load-balance-client/README.md
+++ b/docs/en/setup/examples/grpc-load-balance-client/README.md
@@ -1,4 +1,4 @@
-# GRPC load balance client
+# GRPC load balance Client
GRPC client support connect to multiple server address, and use `round-robin` policy for load-balance server before send each request.
diff --git a/docs/en/setup/examples/high-throughput-queue/README.md b/docs/en/setup/examples/high-throughput-queue/README.md
new file mode 100644
index 0000000..a403b7f
--- /dev/null
+++ b/docs/en/setup/examples/high-throughput-queue/README.md
@@ -0,0 +1,19 @@
+# High Throughput Queue
+
+High throughput queues can send messages to multiple channels by using `round-robin` policy for load-balance, and consume messages from each queue in parallel.
+This prevents the single-threaded sender from performing network IO, blocking the `receiver`/`fetcher` to the queue.
+
+## Config
+
+Both existing two queues(`memory`, `mmap`) support this feature, add the `partition` to the queue config, default is `1`.
+
+```yaml
+queue:
+ plugin_name: "memory-queue"
+ # The maximum buffer event size.
+ event_buffer_size: ${SATELLITE_QUEUE_EVENT_BUFFER_SIZE:5000}
+ # The partition count of queue.
+ partition: ${SATELLITE_QUEUE_PARTITION:2}
+```
+
+Following the config, we create a partitioned queue, have two sub-queue, each sub-queue has 5000 buffer.
diff --git a/docs/en/setup/examples/transmit-log-to-kafka/README.md b/docs/en/setup/examples/transmit-log-to-kafka/README.md
index c555e11..196d5b7 100644
--- a/docs/en/setup/examples/transmit-log-to-kafka/README.md
+++ b/docs/en/setup/examples/transmit-log-to-kafka/README.md
@@ -1,4 +1,4 @@
-# Transmit protocol from agent
+# Transmit Log to Kafka
Using Satellite to receive the SkyWalking log protocol from agent, and transport data to the Kafka Topic.
diff --git a/docs/en/setup/plugins/plugin-list.md b/docs/en/setup/plugins/plugin-list.md
index b34cb6d..76aa5ca 100755
--- a/docs/en/setup/plugins/plugin-list.md
+++ b/docs/en/setup/plugins/plugin-list.md
@@ -1,47 +1,47 @@
# Plugin List
- Client
- - [grpc-client](./client_grpc-client.md)
- - [kafka-client](./client_kafka-client.md)
+ - [GRPC Client](./client_grpc-client.md)
+ - [Kafka Client](./client_kafka-client.md)
- Fallbacker
- - [none-fallbacker](./fallbacker_none-fallbacker.md)
- - [timer-fallbacker](./fallbacker_timer-fallbacker.md)
+ - [None Fallbacker](./fallbacker_none-fallbacker.md)
+ - [Timer Fallbacker](./fallbacker_timer-fallbacker.md)
- Fetcher
- - [prometheus-metrics-fetcher](./fetcher_prometheus-metrics-fetcher.md)
+ - [Prometheus Metrics Fetcher](./fetcher_prometheus-metrics-fetcher.md)
- Filter
- 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)
- - [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)
+ - [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)
+ - [Native CDS GRPC Forwarder](./forwarder_nativecds-grpc-forwarder.md)
+ - [Native Event GRPC Forwarder](./forwarder_nativeevent-grpc-forwarder.md)
+ - [Native JVM GRPC Forwarder](./forwarder_nativejvm-grpc-forwarder.md)
+ - [Native Log GRPC Forwarder](./forwarder_nativelog-grpc-forwarder.md)
+ - [Native Log Kafka Forwarder](./forwarder_nativelog-kafka-forwarder.md)
+ - [Native Management GRPC Forwarder](./forwarder_nativemanagement-grpc-forwarder.md)
+ - [Native Meter GRPC Forwarder](./forwarder_nativemeter-grpc-forwarder.md)
+ - [Native Profile GRPC Forwarder](./forwarder_nativeprofile-grpc-forwarder.md)
+ - [Native Tracing GRPC Forwarder](./forwarder_nativetracing-grpc-forwarder.md)
- Parser
- Queue
- - [memory-queue](./queue_memory-queue.md)
- - [mmap-queue](./queue_mmap-queue.md)
- - [none-queue](./queue_none-queue.md)
+ - [Memory Queue](./queue_memory-queue.md)
+ - [Mmap Queue](./queue_mmap-queue.md)
+ - [None Queue](./queue_none-queue.md)
- 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)
- - [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)
+ - [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 Native CDS Receiver](./receiver_grpc-nativecds-receiver.md)
+ - [GRPC Native Event Receiver](./receiver_grpc-nativeevent-receiver.md)
+ - [GRPC Native JVM Receiver](./receiver_grpc-nativejvm-receiver.md)
+ - [GRPC Native Log Receiver](./receiver_grpc-nativelog-receiver.md)
+ - [GRPC Native Management Receiver](./receiver_grpc-nativemanagement-receiver.md)
+ - [GRPC Native Meter Receiver](./receiver_grpc-nativemeter-receiver.md)
+ - [GRPC Native Profile Receiver](./receiver_grpc-nativeprofile-receiver.md)
+ - [GRPC Native Tracing Receiver](./receiver_grpc-nativetracing-receiver.md)
+ - [HTTP Native Log Receiver](./receiver_http-nativelog-receiver.md)
- Server
- - [grpc-server](./server_grpc-server.md)
- - [http-server](./server_http-server.md)
- - [prometheus-server](./server_prometheus-server.md)
+ - [GRPC Server](./server_grpc-server.md)
+ - [HTTP Server](./server_http-server.md)
+ - [Prometheus Server](./server_prometheus-server.md)
diff --git a/docs/en/setup/plugins/queue_memory-queue.md b/docs/en/setup/plugins/queue_memory-queue.md
index 224debe..a175cb2 100755
--- a/docs/en/setup/plugins/queue_memory-queue.md
+++ b/docs/en/setup/plugins/queue_memory-queue.md
@@ -5,9 +5,13 @@ This is a memory queue to buffer the input event.
```yaml
# The maximum buffer event size.
event_buffer_size: 5000
+
+# The partition count of queue.
+partition: 1
```
## Configuration
|Name|Type|Description|
|----|----|-----------|
| event_buffer_size | int | configThe maximum buffer event size. |
+| partition | int | The total partition count. |
diff --git a/docs/en/setup/plugins/queue_mmap-queue.md b/docs/en/setup/plugins/queue_mmap-queue.md
index 2c7d7fc..9990bfb 100755
--- a/docs/en/setup/plugins/queue_mmap-queue.md
+++ b/docs/en/setup/plugins/queue_mmap-queue.md
@@ -15,6 +15,9 @@ flush_period: 1000
flush_ceiling_num: 10000
# The max size of the input event. Default value is 20k.
max_event_size: 20480
+
+# The partition count of queue.
+partition: 1
```
## Configuration
|Name|Type|Description|
@@ -25,4 +28,5 @@ max_event_size: 20480
| flush_period | int | The period flush time. The unit is ms. |
| flush_ceiling_num | int | The max number in one flush time. |
| max_event_size | int | The max size of the input event. |
+| partition | int | The total partition count. |
diff --git a/docs/en/setup/plugins/queue_none-queue.md b/docs/en/setup/plugins/queue_none-queue.md
index 890df50..7c19703 100755
--- a/docs/en/setup/plugins/queue_none-queue.md
+++ b/docs/en/setup/plugins/queue_none-queue.md
@@ -2,8 +2,12 @@
## Description
This is an empty queue for direct connection protocols, such as SkyWalking native configuration discovery service protocol.
## DefaultConfig
-```yaml```
+```yaml
+# The partition count of queue.
+partition: 1
+```
## Configuration
|Name|Type|Description|
|----|----|-----------|
+| partition | int | The total partition count. |
diff --git a/docs/menu.yml b/docs/menu.yml
index 8d46699..79ace79 100644
--- a/docs/menu.yml
+++ b/docs/menu.yml
@@ -21,7 +21,7 @@ catalog:
catalog:
- name: What is SkyWalking Satellite?
catalog:
- - name: Overview and Core concepts
+ - name: Overview and Core Concepts
path: /en/concepts-and-designs/overview
- name: Project Goals
path: /en/concepts-and-designs/project-goals
@@ -29,11 +29,11 @@ catalog:
catalog:
- name: Module Design
path: /en/concepts-and-designs/module_design
- - name: Plugin mechanism
+ - name: Plugin Mechanism
path: /en/concepts-and-designs/plugin_mechanism
- name: Project Structure
path: /en/concepts-and-designs/project_structue
- - name: The design of the memory mapped queue
+ - name: Memory mapped Queue
path: /en/concepts-and-designs/mmap-queue
- name: Setup
catalog:
@@ -49,113 +49,115 @@ catalog:
path: /en/setup/configuration/override-settings
- name: Examples
catalog:
- - name: Transmit protocol from agent
- path: /en/setup/examples/transmit-protocol-from-agent/readme
- - name: Transmit log to kafka
+ - name: Forward Agent Data to OAP Server
+ path: /en/setup/examples/forward-agent-data-to-oap-server/readme
+ - name: Transmit Log to Kafka
path: /en/setup/examples/transmit-log-to-kafka/readme
- - name: GRPC load balance client
+ - name: GRPC load balance Client
path: /en/setup/examples/grpc-load-balance-client/readme
+ - name: High Throughput Queue
+ path: /en/setup/examples/high-throughput-queue/readme
- name: Plugins
catalog:
- - name: client
+ - name: Client
catalog:
- - name: grpc client
+ - name: GRPC Client
path: /en/setup/plugins/client_grpc-client
- - name: kafka client
+ - name: Kafka Client
path: /en/setup/plugins/client_kafka-client
- - name: fallbacker
+ - name: Fallbacker
catalog:
- - name: none fallbacker
+ - name: None Fallbacker
path: /en/setup/plugins/fallbacker_none-fallbacker
- - name: timer fallbacker
+ - name: Timer Fallbacker
path: /en/setup/plugins/fallbacker_timer-fallbacker
- - name: fetcher
+ - name: Fetcher
catalog:
- - name: prometheus metrics fetcher
+ - name: Prometheus Metrics Fetcher
path: /en/setup/plugins/fetcher_prometheus-metrics-fetcher
- - name: forwarder
+ - name: Forwarder
catalog:
- - name: envoy als v2 grpc forwarder
+ - name: Envoy ALS v2 GRPC Forwarder
path: /en/setup/plugins/forwarder_envoy-als-v2-grpc-forwarder
- - name: envoy als v3 grpc forwarder
+ - name: Envoy ALS v3 GRPC Forwarder
path: /en/setup/plugins/forwarder_envoy-als-v3-grpc-forwarder
- - name: envoy metrics v2 grpc forwarder
+ - name: Envoy Metrics v2 GRPC Forwarder
path: /en/setup/plugins/forwarder_envoy-metrics-v2-grpc-forwarder
- - name: envoy metrics v3 grpc forwarder
+ - name: Envoy Metrics v3 GRPC Forwarder
path: /en/setup/plugins/forwarder_envoy-metrics-v3-grpc-forwarder
- - name: nativecds grpc forwarder
+ - name: Native CDS GRPC Forwarder
path: /en/setup/plugins/forwarder_nativecds-grpc-forwarder
- - name: nativeevent grpc forwarder
+ - name: Native Event GRPC Forwarder
path: /en/setup/plugins/forwarder_nativeevent-grpc-forwarder
- - name: nativejvm grpc forwarder
+ - name: Native JVM GRPC Forwarder
path: /en/setup/plugins/forwarder_nativejvm-grpc-forwarder
- - name: nativelog grpc forwarder
+ - name: Native Log GRPC Forwarder
path: /en/setup/plugins/forwarder_nativelog-grpc-forwarder
- - name: nativelog kafka forwarder
+ - name: Native Log Kafka Forwarder
path: /en/setup/plugins/forwarder_nativelog-kafka-forwarder
- - name: nativemanagement grpc forwarder
+ - name: Native Management GRPC Forwarder
path: /en/setup/plugins/forwarder_nativemanagement-grpc-forwarder
- - name: nativemeter grpc forwarder
+ - name: Native Meter GRPC Forwarder
path: /en/setup/plugins/forwarder_nativemeter-grpc-forwarder
- - name: nativeprofile grpc forwarder
+ - name: Native Profile GRPC Forwarder
path: /en/setup/plugins/forwarder_nativeprofile-grpc-forwarder
- - name: nativetracing grpc forwarder
+ - name: Native Tracing GRPC Forwarder
path: /en/setup/plugins/forwarder_nativetracing-grpc-forwarder
- - name: queue
+ - name: Queue
catalog:
- - name: memory queue
+ - name: Memory Queue
path: /en/setup/plugins/queue_memory-queue
- - name: mmap queue
+ - name: Mmap Queue
path: /en/setup/plugins/queue_mmap-queue
- - name: none queue
+ - name: None Queue
path: /en/setup/plugins/queue_none-queue
- - name: receiver
+ - name: Receiver
catalog:
- - name: grpc envoy als v2 receiver
+ - name: GRPC Envoy ALS v2 Receiver
path: /en/setup/plugins/receiver_grpc-envoy-als-v2-receiver
- - name: grpc envoy als v3 receiver
+ - name: GRPC Envoy ALS v3 Receiver
path: /en/setup/plugins/receiver_grpc-envoy-als-v3-receiver
- - name: grpc envoy metrics v2 receiver
+ - name: GRPC Envoy Metrics v2 Receiver
path: /en/setup/plugins/receiver_grpc-envoy-metrics-v2-receiver
- - name: grpc envoy metrics v3 receiver
+ - name: GRPC Envoy Metrics v3 Receiver
path: /en/setup/plugins/receiver_grpc-envoy-metrics-v3-receiver
- - name: grpc nativecds receiver
+ - name: GRPC Native CDS Receiver
path: /en/setup/plugins/receiver_grpc-nativecds-receiver
- - name: grpc nativeevent receiver
+ - name: GRPC Native Event Receiver
path: /en/setup/plugins/receiver_grpc-nativeevent-receiver
- - name: grpc nativejvm receiver
+ - name: GRPC Native JVM Receiver
path: /en/setup/plugins/receiver_grpc-nativejvm-receiver
- - name: grpc nativelog receiver
+ - name: GRPC Native Log Receiver
path: /en/setup/plugins/receiver_grpc-nativelog-receiver
- - name: grpc nativemanagement receiver
+ - name: GRPC Native Management Receiver
path: /en/setup/plugins/receiver_grpc-nativemanagement-receiver
- - name: grpc nativemeter receiver
+ - name: GRPC Native Meter Receiver
path: /en/setup/plugins/receiver_grpc-nativemeter-receiver
- - name: grpc nativeprofile receiver
+ - name: GRPC Native Profile Receiver
path: /en/setup/plugins/receiver_grpc-nativeprofile-receiver
- - name: grpc nativetracing receiver
+ - name: GRPC Native Tracing Receiver
path: /en/setup/plugins/receiver_grpc-nativetracing-receiver
- - name: http nativelog receiver
+ - name: HTTP Native Log Receiver
path: /en/setup/plugins/receiver_http-nativelog-receiver
- - name: server
+ - name: Server
catalog:
- - name: grpc server
+ - name: GRPC Server
path: /en/setup/plugins/server_grpc-server
- - name: http server
+ - name: HTTP Server
path: /en/setup/plugins/server_http-server
- - name: prometheus server
+ - name: Prometheus Server
path: /en/setup/plugins/server_prometheus-server
- name: Guides
catalog:
- name: Contribution
catalog:
- - name: How to write a plugin
+ - name: How to write a Plugin
path: /en/guides/contribution/How-to-write-plugin
- - name: How to release
+ - name: How to Release
path: /en/guides/contribution/How-to-release
- name: Test
catalog:
- - name: How to do unit test
+ - name: How to do Unit Test
path: /en/guides/test/How-to-unit-test
- name: Compile
catalog:
diff --git a/internal/pkg/plugin/definition.go b/internal/pkg/plugin/definition.go
index fd7c278..f38e3a2 100644
--- a/internal/pkg/plugin/definition.go
+++ b/internal/pkg/plugin/definition.go
@@ -24,6 +24,8 @@ const NameField = "plugin_name"
type Plugin interface {
// Name returns the name of the specific plugin.
Name() string
+ // ShowName returns the specific name show on documentations.
+ ShowName() string
// Description returns the description of the specific plugin.
Description() string
// DefaultConfig returns the default config, that is a YAML pattern.
diff --git a/internal/pkg/plugin/plugins/democategory/demodir1/demodir2/plugin_test.go b/internal/pkg/plugin/plugins/democategory/demodir1/demodir2/plugin_test.go
index b30b062..42387cf 100644
--- a/internal/pkg/plugin/plugins/democategory/demodir1/demodir2/plugin_test.go
+++ b/internal/pkg/plugin/plugins/democategory/demodir1/demodir2/plugin_test.go
@@ -49,6 +49,10 @@ func (d *DemoPlugin) Name() string {
return "demo-plugin"
}
+func (d *DemoPlugin) ShowName() string {
+ return "Demo Plugin"
+}
+
func (d *DemoPlugin) Description() string {
return "this is just a demo"
}
diff --git a/internal/pkg/plugin/registry.go b/internal/pkg/plugin/registry.go
index 6f1b7ae..e6a3019 100644
--- a/internal/pkg/plugin/registry.go
+++ b/internal/pkg/plugin/registry.go
@@ -79,7 +79,7 @@ func Get(category reflect.Type, cfg Config) Plugin {
t = t.Elem()
}
plugin := reflect.New(t).Interface().(Plugin)
- initializing(plugin, cfg)
+ Initializing(plugin, cfg)
return plugin
}
@@ -96,8 +96,8 @@ func nameFinder(cfg interface{}) string {
return name.(string)
}
-// initializing initialize the fields by fields mapping.
-func initializing(plugin Plugin, cfg Config) {
+// Initializing initialize the fields by fields mapping.
+func Initializing(plugin Plugin, cfg Config) {
v := viper.New()
v.SetConfigType("yaml")
if plugin.DefaultConfig() != "" {
diff --git a/internal/satellite/config/loader_test.go b/internal/satellite/config/loader_test.go
index 002a606..d2492af 100644
--- a/internal/satellite/config/loader_test.go
+++ b/internal/satellite/config/loader_test.go
@@ -137,6 +137,7 @@ func pipes() []*PipeConfig {
"commonfields_pipe_name": "logpipe",
"plugin_name": "memory-queue",
"event_buffer_size": 5000,
+ "partition": 1,
},
},
Processor: &processor.ProcessorConfig{
diff --git a/internal/satellite/event/event.go b/internal/satellite/event/event.go
index 97e1240..d8b9c4b 100644
--- a/internal/satellite/event/event.go
+++ b/internal/satellite/event/event.go
@@ -26,7 +26,10 @@ import (
type Type int32
// Offset is a generic form, which allows having different definitions in different Queues.
-type Offset string
+type Offset struct {
+ Partition int
+ Position string
+}
// BatchEvents is used by Forwarder to forward.
type BatchEvents []*v1.SniffData
@@ -34,7 +37,7 @@ type BatchEvents []*v1.SniffData
// OutputEventContext is a container to store the output context.
type OutputEventContext struct {
Context map[string]*v1.SniffData
- Offset Offset
+ Offset *Offset
}
// Put puts the incoming event into the context.
@@ -51,3 +54,10 @@ func (c *OutputEventContext) Get(eventName string) (*v1.SniffData, error) {
}
return e, nil
}
+
+func (o *Offset) String() string {
+ if o == nil {
+ return ""
+ }
+ return fmt.Sprintf("%d_%s", o.Partition, o.Position)
+}
diff --git a/internal/satellite/module/buffer/buffer.go b/internal/satellite/module/buffer/buffer.go
index 6fc1569..69c8824 100644
--- a/internal/satellite/module/buffer/buffer.go
+++ b/internal/satellite/module/buffer/buffer.go
@@ -25,8 +25,8 @@ import (
// BatchBuffer is a buffer to cache the input data in Sender.
type BatchBuffer struct {
buf []*event.OutputEventContext // cache
- first event.Offset // the first OutputEventContext offset
- last event.Offset // the last OutputEventContext offset
+ first *event.Offset // the first OutputEventContext offset
+ last *event.Offset // the last OutputEventContext offset
size int // usage size
cap int // the max capacity
}
@@ -35,8 +35,8 @@ type BatchBuffer struct {
func NewBatchBuffer(capacity int) *BatchBuffer {
return &BatchBuffer{
buf: make([]*event.OutputEventContext, capacity),
- first: "",
- last: "",
+ first: nil,
+ last: nil,
size: 0,
cap: capacity,
}
@@ -48,12 +48,12 @@ func (b *BatchBuffer) Buf() []*event.OutputEventContext {
}
// First returns the first OutputEventContext offset.
-func (b *BatchBuffer) First() event.Offset {
+func (b *BatchBuffer) First() *event.Offset {
return b.first
}
// Last returns the last OutputEventContext offset.
-func (b *BatchBuffer) Last() event.Offset {
+func (b *BatchBuffer) Last() *event.Offset {
return b.last
}
@@ -67,15 +67,14 @@ func (b *BatchBuffer) Add(data *event.OutputEventContext) {
if b.size == b.cap {
log.Logger.Errorf("cannot add one item to the fulling BatchBuffer, the capacity is %d", b.cap)
return
- } else if data.Offset == "" {
+ } else if data.Offset == nil {
log.Logger.Errorf("cannot add one item to BatchBuffer because the input data is illegal, the offset is empty")
return
}
if b.size == 0 {
b.first = data.Offset
- } else {
- b.last = data.Offset
}
+ b.last = data.Offset
b.buf[b.size] = data
b.size++
}
diff --git a/internal/satellite/module/buffer/buffer_test.go b/internal/satellite/module/buffer/buffer_test.go
index 953be39..6dde07b 100644
--- a/internal/satellite/module/buffer/buffer_test.go
+++ b/internal/satellite/module/buffer/buffer_test.go
@@ -33,27 +33,27 @@ func TestNewBuffer(t *testing.T) {
}{
{
name: "add-1",
- args: &event.OutputEventContext{Offset: "1"},
+ args: &event.OutputEventContext{Offset: &event.Offset{Position: "1"}},
want: 1,
},
{
name: "add-2",
- args: &event.OutputEventContext{Offset: "1"},
+ args: &event.OutputEventContext{Offset: &event.Offset{Position: "1"}},
want: 2,
},
{
name: "add-3",
- args: &event.OutputEventContext{Offset: "1"},
+ args: &event.OutputEventContext{Offset: &event.Offset{Position: "1"}},
want: 3,
},
{
name: "add-4",
- args: &event.OutputEventContext{Offset: "1"},
+ args: &event.OutputEventContext{Offset: &event.Offset{Position: "1"}},
want: 3,
},
{
name: "add-5",
- args: &event.OutputEventContext{Offset: "1"},
+ args: &event.OutputEventContext{Offset: &event.Offset{Position: "1"}},
want: 3,
},
}
diff --git a/internal/satellite/module/gatherer/api/gatherer.go b/internal/satellite/module/gatherer/api/gatherer.go
index af4fb7b..19bdd52 100644
--- a/internal/satellite/module/gatherer/api/gatherer.go
+++ b/internal/satellite/module/gatherer/api/gatherer.go
@@ -26,10 +26,12 @@ import (
// Gatherer is the APM data collection module in Satellite.
type Gatherer interface {
api.Module
+ // PartitionCount is the all partition counter of gatherer. All event is partitioned.
+ PartitionCount() int
// OutputDataChannel is a blocking channel to transfer the apm data to the upstream processor module.
- OutputDataChannel() <-chan *queue.SequenceEvent
+ OutputDataChannel(partition int) <-chan *queue.SequenceEvent
// Ack the sent offset.
- Ack(lastOffset event.Offset)
+ Ack(lastOffset *event.Offset)
// Inject the Processor module.
SetProcessor(processor api.Module) error
}
diff --git a/internal/satellite/module/gatherer/create.go b/internal/satellite/module/gatherer/create.go
index 8973525..81a1f29 100644
--- a/internal/satellite/module/gatherer/create.go
+++ b/internal/satellite/module/gatherer/create.go
@@ -21,7 +21,7 @@ import (
"github.com/apache/skywalking-satellite/internal/satellite/module/gatherer/api"
"github.com/apache/skywalking-satellite/internal/satellite/sharing"
fetcher "github.com/apache/skywalking-satellite/plugins/fetcher/api"
- queue "github.com/apache/skywalking-satellite/plugins/queue/api"
+ "github.com/apache/skywalking-satellite/plugins/queue/partition"
receiver "github.com/apache/skywalking-satellite/plugins/receiver/api"
server "github.com/apache/skywalking-satellite/plugins/server/api"
)
@@ -40,9 +40,8 @@ func NewGatherer(cfg *api.GathererConfig) api.Gatherer {
func newFetcherGatherer(cfg *api.GathererConfig) *FetcherGatherer {
return &FetcherGatherer{
config: cfg,
- runningQueue: queue.GetQueue(cfg.QueueConfig),
+ runningQueue: partition.NewPartitionQueue(cfg.QueueConfig),
runningFetcher: fetcher.GetFetcher(cfg.FetcherConfig),
- outputChannel: make(chan *queue.SequenceEvent),
}
}
@@ -50,9 +49,8 @@ func newFetcherGatherer(cfg *api.GathererConfig) *FetcherGatherer {
func newReceiverGatherer(cfg *api.GathererConfig) *ReceiverGatherer {
return &ReceiverGatherer{
config: cfg,
- runningQueue: queue.GetQueue(cfg.QueueConfig),
+ runningQueue: partition.NewPartitionQueue(cfg.QueueConfig),
runningReceiver: receiver.GetReceiver(cfg.ReceiverConfig),
runningServer: sharing.Manager[cfg.ServerName].(server.Server),
- outputChannel: make(chan *queue.SequenceEvent),
}
}
diff --git a/internal/satellite/module/gatherer/fetcher_gatherer.go b/internal/satellite/module/gatherer/fetcher_gatherer.go
index ab68276..41812e9 100644
--- a/internal/satellite/module/gatherer/fetcher_gatherer.go
+++ b/internal/satellite/module/gatherer/fetcher_gatherer.go
@@ -31,6 +31,7 @@ import (
"github.com/apache/skywalking-satellite/internal/satellite/telemetry"
fetcher "github.com/apache/skywalking-satellite/plugins/fetcher/api"
queue "github.com/apache/skywalking-satellite/plugins/queue/api"
+ "github.com/apache/skywalking-satellite/plugins/queue/partition"
)
type FetcherGatherer struct {
@@ -39,10 +40,10 @@ type FetcherGatherer struct {
// dependency plugins
runningFetcher fetcher.Fetcher
- runningQueue queue.Queue
+ runningQueue *partition.PartitionedQueue
// self components
- outputChannel chan *queue.SequenceEvent
+ outputChannel []chan *queue.SequenceEvent
// metrics
fetchCounter *telemetry.Counter
@@ -58,6 +59,10 @@ func (f *FetcherGatherer) Prepare() error {
log.Logger.WithField("pipe", f.config.PipeName).Infof("the %s queue failed when initializing", f.runningQueue.Name())
return err
}
+ f.outputChannel = make([]chan *queue.SequenceEvent, f.runningQueue.TotalPartitionCount())
+ for p := 0; p < f.runningQueue.TotalPartitionCount(); p++ {
+ f.outputChannel[p] = make(chan *queue.SequenceEvent)
+ }
f.fetchCounter = telemetry.NewCounter("gatherer_fetch_count", "Total number of the receiving count in the Gatherer.", "pipe", "status")
f.queueOutputCounter = telemetry.NewCounter("queue_output_count", "Total number of the output count in the Queue of Gatherer.", "pipe", "status")
return nil
@@ -66,7 +71,7 @@ func (f *FetcherGatherer) Prepare() error {
func (f *FetcherGatherer) Boot(ctx context.Context) {
log.Logger.WithField("pipe", f.config.PipeName).Info("fetch_gatherer module is starting...")
var wg sync.WaitGroup
- wg.Add(2)
+ wg.Add(f.PartitionCount() + 1)
go func() {
defer wg.Done()
childCtx, cancel := context.WithCancel(ctx)
@@ -87,6 +92,13 @@ func (f *FetcherGatherer) Boot(ctx context.Context) {
}
}()
+ for p := 0; p < f.PartitionCount(); p++ {
+ f.consumeQueue(ctx, p, &wg)
+ }
+ wg.Wait()
+}
+
+func (f *FetcherGatherer) consumeQueue(ctx context.Context, p int, wg *sync.WaitGroup) {
go func() {
defer wg.Done()
childCtx, cancel := context.WithCancel(ctx)
@@ -97,8 +109,8 @@ func (f *FetcherGatherer) Boot(ctx context.Context) {
f.Shutdown()
return
default:
- if e, err := f.runningQueue.Dequeue(); err == nil {
- f.outputChannel <- e
+ if e, err := f.runningQueue.Dequeue(p); err == nil {
+ f.outputChannel[p] <- e
f.queueOutputCounter.Inc(f.config.PipeName, "success")
} else if err == queue.ErrEmpty {
time.Sleep(time.Second)
@@ -120,11 +132,15 @@ func (f *FetcherGatherer) Shutdown() {
}
}
-func (f *FetcherGatherer) OutputDataChannel() <-chan *queue.SequenceEvent {
- return f.outputChannel
+func (f *FetcherGatherer) PartitionCount() int {
+ return len(f.outputChannel)
+}
+
+func (f *FetcherGatherer) OutputDataChannel(index int) <-chan *queue.SequenceEvent {
+ return f.outputChannel[index]
}
-func (f *FetcherGatherer) Ack(lastOffset event.Offset) {
+func (f *FetcherGatherer) Ack(lastOffset *event.Offset) {
f.runningQueue.Ack(lastOffset)
}
diff --git a/internal/satellite/module/gatherer/receiver_gatherer.go b/internal/satellite/module/gatherer/receiver_gatherer.go
index d0dfc73..e215bb0 100644
--- a/internal/satellite/module/gatherer/receiver_gatherer.go
+++ b/internal/satellite/module/gatherer/receiver_gatherer.go
@@ -32,6 +32,7 @@ import (
processor "github.com/apache/skywalking-satellite/internal/satellite/module/processor/api"
"github.com/apache/skywalking-satellite/internal/satellite/telemetry"
queue "github.com/apache/skywalking-satellite/plugins/queue/api"
+ "github.com/apache/skywalking-satellite/plugins/queue/partition"
receiver "github.com/apache/skywalking-satellite/plugins/receiver/api"
server "github.com/apache/skywalking-satellite/plugins/server/api"
@@ -44,11 +45,11 @@ type ReceiverGatherer struct {
// dependency plugins
runningReceiver receiver.Receiver
- runningQueue queue.Queue
+ runningQueue *partition.PartitionedQueue
runningServer server.Server
// self components
- outputChannel chan *queue.SequenceEvent
+ outputChannel []chan *queue.SequenceEvent
// metrics
receiveCounter *telemetry.Counter
queueOutputCounter *telemetry.Counter
@@ -64,6 +65,10 @@ func (r *ReceiverGatherer) Prepare() error {
log.Logger.WithField("pipe", r.config.PipeName).Infof("the %s queue failed when initializing", r.runningQueue.Name())
return err
}
+ r.outputChannel = make([]chan *queue.SequenceEvent, r.runningQueue.TotalPartitionCount())
+ for p := 0; p < r.runningQueue.TotalPartitionCount(); p++ {
+ r.outputChannel[p] = make(chan *queue.SequenceEvent)
+ }
r.receiveCounter = telemetry.NewCounter("gatherer_receive_count", "Total number of the receiving count in the Gatherer.", "pipe", "status")
r.queueOutputCounter = telemetry.NewCounter("queue_output_count", "Total number of the output count in the Queue of Gatherer.", "pipe", "status")
return nil
@@ -72,7 +77,7 @@ func (r *ReceiverGatherer) Prepare() error {
func (r *ReceiverGatherer) Boot(ctx context.Context) {
r.runningReceiver.RegisterSyncInvoker(r)
var wg sync.WaitGroup
- wg.Add(2)
+ wg.Add(r.PartitionCount() + 1)
log.Logger.WithField("pipe", r.config.PipeName).Info("receive_gatherer module is starting...")
go func() {
childCtx, cancel := context.WithCancel(ctx)
@@ -96,6 +101,13 @@ func (r *ReceiverGatherer) Boot(ctx context.Context) {
}
}()
+ for p := 0; p < r.PartitionCount(); p++ {
+ r.consumeQueue(ctx, p, &wg)
+ }
+ wg.Wait()
+}
+
+func (r *ReceiverGatherer) consumeQueue(ctx context.Context, p int, wg *sync.WaitGroup) {
go func() {
childCtx, cancel := context.WithCancel(ctx)
defer wg.Done()
@@ -106,8 +118,8 @@ func (r *ReceiverGatherer) Boot(ctx context.Context) {
r.Shutdown()
return
default:
- if e, err := r.runningQueue.Dequeue(); err == nil {
- r.outputChannel <- e
+ if e, err := r.runningQueue.Dequeue(p); err == nil {
+ r.outputChannel[p] <- e
r.queueOutputCounter.Inc(r.config.PipeName, "success")
} else if err == queue.ErrEmpty {
time.Sleep(time.Second)
@@ -121,7 +133,6 @@ func (r *ReceiverGatherer) Boot(ctx context.Context) {
}
}
}()
- wg.Wait()
}
func (r *ReceiverGatherer) Shutdown() {
@@ -135,11 +146,15 @@ func (r *ReceiverGatherer) Shutdown() {
}
}
-func (r *ReceiverGatherer) OutputDataChannel() <-chan *queue.SequenceEvent {
- return r.outputChannel
+func (r *ReceiverGatherer) PartitionCount() int {
+ return len(r.outputChannel)
+}
+
+func (r *ReceiverGatherer) OutputDataChannel(index int) <-chan *queue.SequenceEvent {
+ return r.outputChannel[index]
}
-func (r *ReceiverGatherer) Ack(lastOffset event.Offset) {
+func (r *ReceiverGatherer) Ack(lastOffset *event.Offset) {
r.runningQueue.Ack(lastOffset)
}
diff --git a/internal/satellite/module/processor/processor.go b/internal/satellite/module/processor/processor.go
index f2d2a4a..48cc4f7 100644
--- a/internal/satellite/module/processor/processor.go
+++ b/internal/satellite/module/processor/processor.go
@@ -54,14 +54,21 @@ func (p *Processor) Prepare() error {
func (p *Processor) Boot(ctx context.Context) {
log.Logger.WithField("pipe", p.config.PipeName).Info("processor module is starting...")
var wg sync.WaitGroup
- wg.Add(1)
+ wg.Add(p.gatherer.PartitionCount())
+ for partition := 0; partition < p.gatherer.PartitionCount(); partition++ {
+ p.processPerPartition(ctx, partition, &wg)
+ }
+ wg.Wait()
+}
+
+func (p *Processor) processPerPartition(ctx context.Context, partition int, wg *sync.WaitGroup) {
go func() {
childCtx, cancel := context.WithCancel(ctx)
defer wg.Done()
for {
select {
// receive the input event from the output channel of the gatherer
- case e := <-p.gatherer.OutputDataChannel():
+ case e := <-p.gatherer.OutputDataChannel(partition):
c := &event.OutputEventContext{
Offset: e.Offset,
Context: make(map[string]*v1.SniffData),
@@ -72,7 +79,7 @@ func (p *Processor) Boot(ctx context.Context) {
f.Process(c)
}
// send the final context that contains many events to the sender.
- p.sender.InputDataChannel() <- c
+ p.sender.InputDataChannel(partition) <- c
case <-childCtx.Done():
cancel()
p.Shutdown()
@@ -80,7 +87,6 @@ func (p *Processor) Boot(ctx context.Context) {
}
}
}()
- wg.Wait()
}
func (p *Processor) Shutdown() {
diff --git a/internal/satellite/module/sender/api/sender.go b/internal/satellite/module/sender/api/sender.go
index 74a125b..b010108 100644
--- a/internal/satellite/module/sender/api/sender.go
+++ b/internal/satellite/module/sender/api/sender.go
@@ -27,7 +27,7 @@ type Sender interface {
api.SyncInvoker
// InputDataChannel is a blocking channel to receive the apm data from the downstream processor module.
- InputDataChannel() chan<- *event.OutputEventContext
+ InputDataChannel(partition int) chan<- *event.OutputEventContext
// Inject the gatherer module.
SetGatherer(g api.Module) error
diff --git a/internal/satellite/module/sender/create.go b/internal/satellite/module/sender/create.go
index c1dc143..0a89a52 100644
--- a/internal/satellite/module/sender/create.go
+++ b/internal/satellite/module/sender/create.go
@@ -18,8 +18,6 @@
package sender
import (
- "github.com/apache/skywalking-satellite/internal/satellite/event"
- "github.com/apache/skywalking-satellite/internal/satellite/module/buffer"
"github.com/apache/skywalking-satellite/internal/satellite/module/sender/api"
"github.com/apache/skywalking-satellite/internal/satellite/sharing"
client "github.com/apache/skywalking-satellite/plugins/client/api"
@@ -34,10 +32,7 @@ func NewSender(cfg *api.SenderConfig) api.Sender {
runningForwarders: []forwarder.Forwarder{},
runningFallbacker: fallbacker.GetFallbacker(cfg.FallbackerConfig),
runningClient: sharing.Manager[cfg.ClientName].(client.Client),
- input: make(chan *event.OutputEventContext),
listener: make(chan client.ClientStatus),
- flushChannel: make(chan *buffer.BatchBuffer, 1),
- buffer: buffer.NewBatchBuffer(cfg.MaxBufferSize),
}
for _, c := range s.config.ForwardersConfig {
s.runningForwarders = append(s.runningForwarders, forwarder.GetForwarder(c))
diff --git a/internal/satellite/module/sender/sender.go b/internal/satellite/module/sender/sender.go
index b50ea5d..964d36d 100644
--- a/internal/satellite/module/sender/sender.go
+++ b/internal/satellite/module/sender/sender.go
@@ -57,17 +57,18 @@ type Sender struct {
gatherer gatherer.Gatherer
// self components
- input chan *event.OutputEventContext // physical input channel
- listener chan client.ClientStatus // client status listener
- flushChannel chan *buffer.BatchBuffer // forwarder flush channel
- buffer *buffer.BatchBuffer // cache the downstream input data
- blocking int32 // the status of input channel
+ inputs []chan *event.OutputEventContext // physical input channel of partitions
+ listener chan client.ClientStatus // client status listener
+ flushChannel []chan *buffer.BatchBuffer // forwarder flush channel
+ buffers []*buffer.BatchBuffer // cache the downstream petitioned input data
+ blocking int32 // the status of input channel
+ shutdownOnce sync.Once
// metrics
sendCounter *telemetry.Counter
}
-// Prepare register the client status listener to the client manager and open input channel.
+// Prepare register the client status listener to the client manager and open partitioned input channel.
func (s *Sender) Prepare() error {
log.Logger.WithField("pipe", s.config.PipeName).Info("sender module is preparing...")
s.runningClient.RegisterListener(s.listener)
@@ -77,25 +78,35 @@ func (s *Sender) Prepare() error {
return err
}
}
+ s.inputs = make([]chan *event.OutputEventContext, s.gatherer.PartitionCount())
+ s.buffers = make([]*buffer.BatchBuffer, s.gatherer.PartitionCount())
+ s.flushChannel = make([]chan *buffer.BatchBuffer, s.gatherer.PartitionCount())
+ for partition := 0; partition < s.gatherer.PartitionCount(); partition++ {
+ s.inputs[partition] = make(chan *event.OutputEventContext)
+ s.buffers[partition] = buffer.NewBatchBuffer(s.config.MaxBufferSize)
+ s.flushChannel[partition] = make(chan *buffer.BatchBuffer)
+ }
s.sendCounter = telemetry.NewCounter("sender_output_count", "Total number of the output count in the Sender.", "pipe", "status", "type")
return nil
}
-// Boot fetches the downstream input data and forward to external services, such as Kafka and OAP receiver.
+// Boot fetches the downstream partitioned input data and forward to external services, such as Kafka and OAP receiver.
func (s *Sender) Boot(ctx context.Context) {
log.Logger.WithField("pipe", s.config.PipeName).Info("sender module is starting...")
var wg sync.WaitGroup
- wg.Add(3)
- go s.store(ctx, &wg)
+ wg.Add(2*s.gatherer.PartitionCount() + 1)
go s.listen(ctx, &wg)
- go s.flush(ctx, &wg)
+ for partition := 0; partition < s.gatherer.PartitionCount(); partition++ {
+ go s.store(ctx, partition, &wg)
+ go s.flush(ctx, partition, &wg)
+ }
wg.Wait()
}
// store data.
// 1. keep fetching the downstream data when client connected, and put it into BatchBuffer.
// 2. When reaches the buffer limit or receives a timer flush signal, and put BatchBuffer into flushChannel.
-func (s *Sender) store(ctx context.Context, wg *sync.WaitGroup) {
+func (s *Sender) store(ctx context.Context, partition int, wg *sync.WaitGroup) {
defer wg.Done()
defer log.Logger.WithField("pipe", s.config.PipeName).Infof("store routine closed")
childCtx, _ := context.WithCancel(ctx) // nolint
@@ -114,18 +125,18 @@ func (s *Sender) store(ctx context.Context, wg *sync.WaitGroup) {
case <-childCtx.Done():
return
case <-timeTicker.C:
- if s.buffer.Len() >= s.config.MinFlushEvents {
- s.flushChannel <- s.buffer
- s.buffer = buffer.NewBatchBuffer(s.config.MaxBufferSize)
+ if s.buffers[partition].Len() >= s.config.MinFlushEvents {
+ s.flushChannel[partition] <- s.buffers[partition]
+ s.buffers[partition] = buffer.NewBatchBuffer(s.config.MaxBufferSize)
}
- case e := <-s.input:
+ case e := <-s.inputs[partition]:
if e == nil {
continue
}
- s.buffer.Add(e)
- if s.buffer.Len() == s.config.MaxBufferSize {
- s.flushChannel <- s.buffer
- s.buffer = buffer.NewBatchBuffer(s.config.MaxBufferSize)
+ s.buffers[partition].Add(e)
+ if s.buffers[partition].Len() == s.config.MaxBufferSize {
+ s.flushChannel[partition] <- s.buffers[partition]
+ s.buffers[partition] = buffer.NewBatchBuffer(s.config.MaxBufferSize)
}
}
}
@@ -154,7 +165,7 @@ func (s *Sender) listen(ctx context.Context, wg *sync.WaitGroup) {
}
// Keep fetching BatchBuffer to forward.
-func (s *Sender) flush(ctx context.Context, wg *sync.WaitGroup) {
+func (s *Sender) flush(ctx context.Context, partition int, wg *sync.WaitGroup) {
defer wg.Done()
defer log.Logger.WithField("pipe", s.config.PipeName).Infof("flush routine closed")
childCtx, _ := context.WithCancel(ctx) // nolint
@@ -163,7 +174,7 @@ func (s *Sender) flush(ctx context.Context, wg *sync.WaitGroup) {
case <-childCtx.Done():
s.Shutdown()
return
- case b := <-s.flushChannel:
+ case b := <-s.flushChannel[partition]:
s.consume(b)
}
}
@@ -171,17 +182,39 @@ func (s *Sender) flush(ctx context.Context, wg *sync.WaitGroup) {
// Shutdown closes the channels and tries to force forward the events in the buffer.
func (s *Sender) Shutdown() {
+ s.shutdownOnce.Do(func() {
+ s.shutdown0()
+ })
+}
+
+func (s *Sender) shutdown0() {
log.Logger.WithField("pipe", s.config.PipeName).Info("sender module is closing")
- close(s.input)
+ for _, in := range s.inputs {
+ close(in)
+ }
+ var wg sync.WaitGroup
+ finished := make(chan struct{}, 1)
+ wg.Add(len(s.flushChannel))
+ for partition := range s.buffers {
+ go func(p int) {
+ defer wg.Done()
+ s.consume(s.buffers[p])
+ }(partition)
+ }
+ go func() {
+ wg.Wait()
+ close(finished)
+ }()
+
ticker := time.NewTicker(module.ShutdownHookTime)
- for {
- select {
- case <-ticker.C:
- s.consume(s.buffer)
- return
- case b := <-s.flushChannel:
- s.consume(b)
+ select {
+ case <-ticker.C:
+ for _, buffer := range s.buffers {
+ s.consume(buffer)
}
+ return
+ case <-finished:
+ return
}
}
@@ -221,8 +254,8 @@ func (s *Sender) consume(batch *buffer.BatchBuffer) {
s.gatherer.Ack(batch.Last())
}
-func (s *Sender) InputDataChannel() chan<- *event.OutputEventContext {
- return s.input
+func (s *Sender) InputDataChannel(partition int) chan<- *event.OutputEventContext {
+ return s.inputs[partition]
}
func (s *Sender) SyncInvoke(d *v1.SniffData) (*v1.SniffData, error) {
diff --git a/internal/satellite/telemetry/gauge.go b/internal/satellite/telemetry/gauge.go
new file mode 100644
index 0000000..d5af811
--- /dev/null
+++ b/internal/satellite/telemetry/gauge.go
@@ -0,0 +1,58 @@
+// 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 telemetry
+
+import "github.com/prometheus/client_golang/prometheus"
+
+type Gauge struct {
+ Collector
+ name string
+ gauge prometheus.GaugeFunc
+}
+
+func NewGauge(name, help string, getter func() float64, labels ...string) *Gauge {
+ lock.Lock()
+ defer lock.Unlock()
+ rebuildName, rebuildLabels := rebuildGaugeNameAndLabels(name, labels...)
+ collector, ok := collectorContainer[rebuildName]
+ if !ok {
+ gauge := &Gauge{
+ name: name,
+ gauge: prometheus.NewGaugeFunc(prometheus.GaugeOpts{
+ Name: name,
+ Help: help,
+ ConstLabels: rebuildLabels,
+ }, getter),
+ }
+ Register(WithMeta(rebuildName, gauge.gauge))
+ collectorContainer[rebuildName] = gauge
+ collector = gauge
+ }
+ return collector.(*Gauge)
+}
+
+func rebuildGaugeNameAndLabels(name string, labels ...string) (gaugeName string, labelsMap map[string]string) {
+ resultName := name
+ resultLabels := make(map[string]string)
+ for inx := 0; inx < len(labels); inx += 2 {
+ resultName += "_" + labels[inx] + "_" + labels[inx+1]
+ resultLabels[labels[inx]] = labels[inx+1]
+ }
+
+ return resultName, resultLabels
+}
diff --git a/internal/satellite/telemetry/timer.go b/internal/satellite/telemetry/timer.go
new file mode 100644
index 0000000..f632c50
--- /dev/null
+++ b/internal/satellite/telemetry/timer.go
@@ -0,0 +1,76 @@
+// 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 telemetry
+
+import (
+ "sync"
+ "time"
+)
+
+var timerLocker sync.Mutex
+
+type Timer struct {
+ Collector
+ name string
+ sumCounter *Counter
+ countCounter *Counter
+}
+
+type TimeRecorder struct {
+ timer *Timer
+ startTime time.Time
+ labelValues []string
+}
+
+// NewCounter create a new counter if no metric with the same name exists.
+func NewTimer(name, help string, labels ...string) *Timer {
+ timerLocker.Lock()
+ defer timerLocker.Unlock()
+
+ collector, ok := collectorContainer[name]
+ if !ok {
+ timer := &Timer{
+ name: name,
+ sumCounter: NewCounter(name+"_sum", help, labels...),
+ countCounter: NewCounter(name+"_count", help, labels...),
+ }
+ collectorContainer[name] = timer
+ collector = timer
+ }
+ return collector.(*Timer)
+}
+
+// Start a new time recorder
+func (c *Timer) Start(labelValues ...string) *TimeRecorder {
+ return &TimeRecorder{
+ timer: c,
+ startTime: time.Now(),
+ labelValues: labelValues,
+ }
+}
+
+// AddTime add a new duration and count
+func (c *Timer) AddTime(t time.Duration, labelValues ...string) {
+ c.sumCounter.Add(float64(t.Milliseconds()), labelValues...)
+ c.countCounter.Inc(labelValues...)
+}
+
+// Stop the time and record the time
+func (c *TimeRecorder) Stop() {
+ c.timer.AddTime(time.Since(c.startTime), c.labelValues...)
+}
diff --git a/internal/satellite/tools/generate_plugin_doc.go b/internal/satellite/tools/generate_plugin_doc.go
index 0fd5e43..4493f18 100644
--- a/internal/satellite/tools/generate_plugin_doc.go
+++ b/internal/satellite/tools/generate_plugin_doc.go
@@ -29,6 +29,8 @@ import (
"sort"
"strings"
+ "github.com/apache/skywalking-satellite/plugins/queue/partition"
+
"github.com/apache/skywalking-satellite/internal/pkg/log"
"github.com/apache/skywalking-satellite/internal/pkg/plugin"
"github.com/apache/skywalking-satellite/plugins"
@@ -48,6 +50,8 @@ const (
markdownSuffix = ".md"
commentPrefix = "/ "
+
+ categoryQueue = "Queue"
)
func GeneratePluginDoc(outputRootPath, menuFilePath, pluginFilePath string) error {
@@ -102,15 +106,15 @@ func updateMenuPluginListDoc(outputRootPath, menuFilePath, pluginFilePath string
// plugin
implements := []*Catalog{}
curPlugin := &Catalog{
- Name: strings.ToLower(category.Name()),
+ Name: strings.Title(strings.ToLower(category.Name())),
}
// all implements
pluginList := getPluginsByCategory(category)
for _, pluginName := range pluginList {
implements = append(implements, &Catalog{
- Name: strings.ReplaceAll(pluginName, "-", " "),
- Path: strings.TrimRight(fmt.Sprintf("%s/%s", pluginFilePath, getPluginDocFileName(category, pluginName)), markdownSuffix),
+ Name: pluginName.showName,
+ Path: strings.TrimRight(fmt.Sprintf("%s/%s", pluginFilePath, getPluginDocFileName(category, pluginName.defineName)), markdownSuffix),
})
}
curPlugin.Catalog = implements
@@ -131,7 +135,7 @@ func generatePluginListDoc(docDir string, categories []reflect.Type) error {
docStr += "- " + category.Name() + lf
pluginList := getPluginsByCategory(category)
for _, pluginName := range pluginList {
- docStr += " - [" + pluginName + "](./" + getPluginDocFileName(category, pluginName) + ")" + lf
+ docStr += " - [" + pluginName.showName + "](./" + getPluginDocFileName(category, pluginName.defineName) + ")" + lf
if err := generatePluginDoc(docDir, category, pluginName); err != nil {
return err
}
@@ -140,15 +144,15 @@ func generatePluginListDoc(docDir string, categories []reflect.Type) error {
return writeDoc([]byte(docStr), fileName)
}
-func generatePluginDoc(docDir string, category reflect.Type, pluginName string) error {
- docFileName := docDir + "/" + getPluginDocFileName(category, pluginName)
- p := plugin.Get(category, plugin.Config{plugin.NameField: pluginName})
- docRes := topLevel + category.Name() + "/" + pluginName + lf
+func generatePluginDoc(docDir string, category reflect.Type, pluginName *pluginNames) error {
+ docFileName := docDir + "/" + getPluginDocFileName(category, pluginName.defineName)
+ p := plugin.Get(category, plugin.Config{plugin.NameField: pluginName.defineName})
+ docRes := topLevel + category.Name() + "/" + pluginName.defineName + lf
docRes += secondLevel + "Description" + lf
docRes += p.Description() + lf
docRes += generateSupportForwarders(category, p)
docRes += secondLevel + "DefaultConfig" + lf
- docRes += yamlQuoteStart + p.DefaultConfig() + yamlQuoteEnd + lf
+ docRes += yamlQuoteStart + generateDefaultConfig(category, p) + yamlQuoteEnd + lf
docRes += secondLevel + "Configuration" + lf
docRes += generateConfiguration(category, p) + lf
return writeDoc([]byte(docRes), docFileName)
@@ -164,6 +168,15 @@ func GetModuleName() string {
return modName
}
+func generateDefaultConfig(category reflect.Type, p plugin.Plugin) string {
+ configs := p.DefaultConfig()
+ if category.Name() == categoryQueue {
+ partitionQueue := &partition.PartitionedQueue{}
+ configs = fmt.Sprintf("%s%s", configs, partitionQueue.DefaultConfig())
+ }
+ return configs
+}
+
func generateConfiguration(category reflect.Type, p plugin.Plugin) string {
var content = ""
@@ -174,6 +187,12 @@ func generateConfiguration(category reflect.Type, p plugin.Plugin) string {
eachConfigurationItem(configurations, "", func(name, dataType, desc string) {
content += fmt.Sprintf("| %s | %s | %s |%s", name, dataType, desc, lf)
})
+ if category.Name() == categoryQueue {
+ configurations := getConfigurations(category, reflect.TypeOf(&partition.PartitionedQueue{}).Elem())
+ eachConfigurationItem(configurations, "", func(name, dataType, desc string) {
+ content += fmt.Sprintf("| %s | %s | %s |%s", name, dataType, desc, lf)
+ })
+ }
return content
}
@@ -247,13 +266,21 @@ func getConfigurations(category, p reflect.Type) []*pluginConfigurationItem {
// parse field to configuration item
func parsePluginConfigurationItem(field *ast.Field, pType reflect.Type) (*pluginConfigurationItem, *pluginChildrenFinder) {
- if field.Names == nil || field.Tag == nil {
+ if field.Tag == nil {
return nil, nil
}
var fieldName = ""
- for _, n := range field.Names {
- fieldName += n.Name
+ if field.Names != nil {
+ for _, n := range field.Names {
+ fieldName += n.Name
+ }
+ } else {
+ expr, ok := field.Type.(*ast.SelectorExpr)
+ if !ok {
+ return nil, nil
+ }
+ fieldName = expr.Sel.Name
}
pluginField, find := pType.FieldByName(fieldName)
@@ -323,13 +350,25 @@ func generateSupportForwarders(category reflect.Type, p plugin.Plugin) string {
return result
}
-func getPluginsByCategory(category reflect.Type) []string {
+type pluginNames struct {
+ defineName string
+ showName string
+}
+
+func getPluginsByCategory(category reflect.Type) []*pluginNames {
mapping := plugin.Reg[category]
- var keys []string
+ var keys []*pluginNames
for k := range mapping {
- keys = append(keys, k)
+ t := mapping[k].Type()
+ if t.Kind() == reflect.Ptr {
+ t = t.Elem()
+ }
+ p := reflect.New(t).Interface().(plugin.Plugin)
+ keys = append(keys, &pluginNames{defineName: k, showName: p.ShowName()})
}
- sort.Strings(keys)
+ sort.Slice(keys, func(i, j int) bool {
+ return keys[i].defineName < keys[j].defineName
+ })
return keys
}
diff --git a/plugins/client/grpc/client.go b/plugins/client/grpc/client.go
index 22e72fd..d946502 100644
--- a/plugins/client/grpc/client.go
+++ b/plugins/client/grpc/client.go
@@ -31,7 +31,10 @@ import (
"github.com/apache/skywalking-satellite/plugins/client/grpc/resolvers"
)
-const Name = "grpc-client"
+const (
+ Name = "grpc-client"
+ ShowName = "GRPC Client"
+)
type Client struct {
config.CommonFields
@@ -58,6 +61,10 @@ func (c *Client) Name() string {
return Name
}
+func (c *Client) ShowName() string {
+ return ShowName
+}
+
func (c *Client) Description() string {
return "The gRPC client is a sharing plugin to keep connection with the gRPC server and delivery the data to it."
}
diff --git a/plugins/client/kafka/client.go b/plugins/client/kafka/client.go
index 7574555..01d84e4 100644
--- a/plugins/client/kafka/client.go
+++ b/plugins/client/kafka/client.go
@@ -29,7 +29,10 @@ import (
"github.com/apache/skywalking-satellite/plugins/client/api"
)
-const Name = "kafka-client"
+const (
+ Name = "kafka-client"
+ ShowName = "Kafka Client"
+)
type Client struct {
config.CommonFields
@@ -62,6 +65,10 @@ func (c *Client) Name() string {
return Name
}
+func (c *Client) ShowName() string {
+ return ShowName
+}
+
func (c *Client) Description() string {
return "The Kafka client is a sharing plugin to keep connection with the Kafka brokers and delivery the data to it."
}
diff --git a/plugins/fallbacker/none/none_fallbacker.go b/plugins/fallbacker/none/none_fallbacker.go
index f61cff1..7ed21b5 100644
--- a/plugins/fallbacker/none/none_fallbacker.go
+++ b/plugins/fallbacker/none/none_fallbacker.go
@@ -23,7 +23,10 @@ import (
"github.com/apache/skywalking-satellite/plugins/forwarder/api"
)
-const Name = "none-fallbacker"
+const (
+ Name = "none-fallbacker"
+ ShowName = "None Fallbacker"
+)
type Fallbacker struct {
config.CommonFields
@@ -33,6 +36,10 @@ func (f *Fallbacker) Name() string {
return Name
}
+func (f *Fallbacker) ShowName() string {
+ return ShowName
+}
+
func (f *Fallbacker) Description() string {
return "The fallbacker would do nothing when facing failure data."
}
diff --git a/plugins/fallbacker/timer/timer_fallbacker.go b/plugins/fallbacker/timer/timer_fallbacker.go
index 6614588..20201fa 100644
--- a/plugins/fallbacker/timer/timer_fallbacker.go
+++ b/plugins/fallbacker/timer/timer_fallbacker.go
@@ -25,7 +25,10 @@ import (
"github.com/apache/skywalking-satellite/plugins/forwarder/api"
)
-const Name = "timer-fallbacker"
+const (
+ Name = "timer-fallbacker"
+ ShowName = "Timer Fallbacker"
+)
// Fallbacker is a timer fallbacker when forward fails.
type Fallbacker struct {
@@ -39,6 +42,10 @@ func (t *Fallbacker) Name() string {
return Name
}
+func (t *Fallbacker) ShowName() string {
+ return ShowName
+}
+
func (t *Fallbacker) Description() string {
return "This is a timer fallback trigger to process the forward failure data."
}
diff --git a/plugins/fetcher/prometheus/fetcher.go b/plugins/fetcher/prometheus/fetcher.go
index 2c5d5fa..863364f 100644
--- a/plugins/fetcher/prometheus/fetcher.go
+++ b/plugins/fetcher/prometheus/fetcher.go
@@ -37,6 +37,7 @@ import (
const (
Name = "prometheus-metrics-fetcher"
+ ShowName = "Prometheus Metrics Fetcher"
eventName = "prometheus-metrics-event"
)
@@ -93,6 +94,10 @@ func (f *Fetcher) Name() string {
return Name
}
+func (f *Fetcher) ShowName() string {
+ return ShowName
+}
+
func (f *Fetcher) Description() string {
return "This is a fetcher for Skywalking prometheus metrics format, " +
"which will translate Prometheus metrics to Skywalking meter system."
diff --git a/plugins/forwarder/grpc/envoyalsv2/forwarder.go b/plugins/forwarder/grpc/envoyalsv2/forwarder.go
index 04878e1..a5636e5 100644
--- a/plugins/forwarder/grpc/envoyalsv2/forwarder.go
+++ b/plugins/forwarder/grpc/envoyalsv2/forwarder.go
@@ -33,7 +33,10 @@ import (
"github.com/apache/skywalking-satellite/internal/satellite/event"
)
-const Name = "envoy-als-v2-grpc-forwarder"
+const (
+ Name = "envoy-als-v2-grpc-forwarder"
+ ShowName = "Envoy ALS v2 GRPC Forwarder"
+)
type Forwarder struct {
config.CommonFields
@@ -44,6 +47,10 @@ func (f *Forwarder) Name() string {
return Name
}
+func (f *Forwarder) ShowName() string {
+ return ShowName
+}
+
func (f *Forwarder) Description() string {
return "This is a synchronization ALS v2 grpc forwarder with the Envoy ALS protocol."
}
diff --git a/plugins/forwarder/grpc/envoyalsv3/forwarder.go b/plugins/forwarder/grpc/envoyalsv3/forwarder.go
index 53f972f..53245fb 100644
--- a/plugins/forwarder/grpc/envoyalsv3/forwarder.go
+++ b/plugins/forwarder/grpc/envoyalsv3/forwarder.go
@@ -23,6 +23,8 @@ import (
"io"
"reflect"
+ "github.com/apache/skywalking-satellite/internal/satellite/telemetry"
+
v3 "skywalking.apache.org/repo/goapi/proto/envoy/service/accesslog/v3"
v1 "skywalking.apache.org/repo/goapi/satellite/data/v1"
@@ -33,17 +35,48 @@ import (
"github.com/apache/skywalking-satellite/internal/satellite/event"
)
-const Name = "envoy-als-v3-grpc-forwarder"
+const (
+ Name = "envoy-als-v3-grpc-forwarder"
+ ShowName = "Envoy ALS v3 GRPC Forwarder"
+)
type Forwarder struct {
config.CommonFields
alsClient v3.AccessLogServiceClient
+
+ eventReadySendCount *telemetry.Counter
+ eventSendFinishedCount *telemetry.Counter
+ messageReadySendCount *telemetry.Counter
+ messageSendFinishedCount *telemetry.Counter
+ streamingReadySendCount *telemetry.Counter
+ streamingSendFinishedCount *telemetry.Counter
+
+ forwardConnectTime *telemetry.Timer
+ forwardSendTime *telemetry.Timer
+ forwardCloseTime *telemetry.Timer
+}
+
+func (f *Forwarder) init() {
+ f.eventReadySendCount = telemetry.NewCounter("als_event_ready_send", "Total count of the ALS event ready send.")
+ f.eventSendFinishedCount = telemetry.NewCounter("als_event_send_finished", "Total count of the ALS event send finished.")
+ f.messageReadySendCount = telemetry.NewCounter("als_message_ready_send", "Total count of the ALS message ready send.")
+ f.messageSendFinishedCount = telemetry.NewCounter("als_message_send_finished", "Total count of the ALS event sned finished.")
+ f.streamingReadySendCount = telemetry.NewCounter("als_streaming_ready_send", "Total count of the ALS streaming ready send.")
+ f.streamingSendFinishedCount = telemetry.NewCounter("als_streaming_send_finished", "Total count of the ALS streaming send finished.")
+
+ f.forwardConnectTime = telemetry.NewTimer("als_forward_connect_time", "Total time of the open ALS streaming.")
+ f.forwardSendTime = telemetry.NewTimer("als_forward_send_time", "Total time of the ALS send message.")
+ f.forwardCloseTime = telemetry.NewTimer("als_forward_close_time", "Total time of the ALS streaming close.")
}
func (f *Forwarder) Name() string {
return Name
}
+func (f *Forwarder) ShowName() string {
+ return ShowName
+}
+
func (f *Forwarder) Description() string {
return "This is a synchronization ALS v3 grpc forwarder with the Envoy ALS protocol."
}
@@ -59,20 +92,40 @@ func (f *Forwarder) Prepare(connection interface{}) error {
f.Name(), reflect.TypeOf(connection).String())
}
f.alsClient = v3.NewAccessLogServiceClient(client)
+ f.init()
return nil
}
func (f *Forwarder) Forward(batch event.BatchEvents) error {
+ f.eventReadySendCount.Add(float64(len(batch)))
+ for _, e := range batch {
+ data, _ := e.GetData().(*v1.SniffData_EnvoyALSV3List)
+ f.streamingReadySendCount.Add(float64(len(data.EnvoyALSV3List.Messages)))
+ for _, message := range data.EnvoyALSV3List.Messages {
+ if message.GetHttpLogs() != nil {
+ f.messageReadySendCount.Add(float64(len(message.GetHttpLogs().LogEntry)))
+ } else if message.GetTcpLogs() != nil {
+ f.messageReadySendCount.Add(float64(len(message.GetTcpLogs().LogEntry)))
+ }
+ }
+ }
+
for _, e := range batch {
data, ok := e.GetData().(*v1.SniffData_EnvoyALSV3List)
if !ok {
continue
}
+ // open stream
+ timeRecord := f.forwardConnectTime.Start()
stream, err := f.alsClient.StreamAccessLogs(context.Background())
+ timeRecord.Stop()
if err != nil {
log.Logger.Errorf("open grpc stream error %v", err)
return err
}
+
+ // send message
+ timeRecord = f.forwardSendTime.Start()
for _, message := range data.EnvoyALSV3List.Messages {
err := stream.Send(message)
if err != nil {
@@ -80,8 +133,20 @@ func (f *Forwarder) Forward(batch event.BatchEvents) error {
f.closeStream(stream)
return err
}
+ if message.GetHttpLogs() != nil {
+ f.messageSendFinishedCount.Add(float64(len(message.GetHttpLogs().LogEntry)))
+ } else if message.GetTcpLogs() != nil {
+ f.messageSendFinishedCount.Add(float64(len(message.GetTcpLogs().LogEntry)))
+ }
}
+ timeRecord.Stop()
+
+ // close stream
+ timeRecord = f.forwardCloseTime.Start()
f.closeStream(stream)
+ timeRecord.Stop()
+ f.eventSendFinishedCount.Inc()
+ f.streamingSendFinishedCount.Add(float64(len(data.EnvoyALSV3List.Messages)))
}
return nil
}
diff --git a/plugins/forwarder/grpc/envoymetricsv2/forwarder.go b/plugins/forwarder/grpc/envoymetricsv2/forwarder.go
index 85b63aa..d426efc 100644
--- a/plugins/forwarder/grpc/envoymetricsv2/forwarder.go
+++ b/plugins/forwarder/grpc/envoymetricsv2/forwarder.go
@@ -33,7 +33,10 @@ import (
"github.com/apache/skywalking-satellite/internal/satellite/event"
)
-const Name = "envoy-metrics-v2-grpc-forwarder"
+const (
+ Name = "envoy-metrics-v2-grpc-forwarder"
+ ShowName = "Envoy Metrics v2 GRPC Forwarder"
+)
type Forwarder struct {
config.CommonFields
@@ -44,6 +47,10 @@ func (f *Forwarder) Name() string {
return Name
}
+func (f *Forwarder) ShowName() string {
+ return ShowName
+}
+
func (f *Forwarder) Description() string {
return "This is a synchronization Metrics v2 grpc forwarder with the Envoy metrics protocol."
}
diff --git a/plugins/forwarder/grpc/envoymetricsv3/forwarder.go b/plugins/forwarder/grpc/envoymetricsv3/forwarder.go
index a267a04..701e4a1 100644
--- a/plugins/forwarder/grpc/envoymetricsv3/forwarder.go
+++ b/plugins/forwarder/grpc/envoymetricsv3/forwarder.go
@@ -33,7 +33,10 @@ import (
"github.com/apache/skywalking-satellite/internal/satellite/event"
)
-const Name = "envoy-metrics-v3-grpc-forwarder"
+const (
+ Name = "envoy-metrics-v3-grpc-forwarder"
+ ShowName = "Envoy Metrics v3 GRPC Forwarder"
+)
type Forwarder struct {
config.CommonFields
@@ -44,6 +47,10 @@ func (f *Forwarder) Name() string {
return Name
}
+func (f *Forwarder) ShowName() string {
+ return ShowName
+}
+
func (f *Forwarder) Description() string {
return "This is a synchronization Metrics v3 grpc forwarder with the Envoy metrics protocol."
}
diff --git a/plugins/forwarder/grpc/nativecds/forwarder.go b/plugins/forwarder/grpc/nativecds/forwarder.go
index ad44022..f8b4b22 100644
--- a/plugins/forwarder/grpc/nativecds/forwarder.go
+++ b/plugins/forwarder/grpc/nativecds/forwarder.go
@@ -32,7 +32,10 @@ import (
"github.com/apache/skywalking-satellite/internal/satellite/event"
)
-const Name = "nativecds-grpc-forwarder"
+const (
+ Name = "nativecds-grpc-forwarder"
+ ShowName = "Native CDS GRPC Forwarder"
+)
type Forwarder struct {
config.CommonFields
@@ -44,6 +47,10 @@ func (f *Forwarder) Name() string {
return Name
}
+func (f *Forwarder) ShowName() string {
+ return ShowName
+}
+
func (f *Forwarder) Description() string {
return "This is a synchronization grpc forwarder with the SkyWalking native Configuration Discovery Service protocol."
}
diff --git a/plugins/forwarder/grpc/nativeevent/sync_forwarder.go b/plugins/forwarder/grpc/nativeevent/sync_forwarder.go
index 1676e15..ac2add9 100644
--- a/plugins/forwarder/grpc/nativeevent/sync_forwarder.go
+++ b/plugins/forwarder/grpc/nativeevent/sync_forwarder.go
@@ -33,7 +33,10 @@ import (
v1 "skywalking.apache.org/repo/goapi/satellite/data/v1"
)
-const Name = "nativeevent-grpc-forwarder"
+const (
+ Name = "nativeevent-grpc-forwarder"
+ ShowName = "Native Event GRPC Forwarder"
+)
type Forwarder struct {
config.CommonFields
@@ -44,6 +47,10 @@ func (f *Forwarder) Name() string {
return Name
}
+func (f *Forwarder) ShowName() string {
+ return ShowName
+}
+
func (f *Forwarder) Description() string {
return "This is a synchronization grpc forwarder with the SkyWalking native event protocol."
}
diff --git a/plugins/forwarder/grpc/nativejvm/sync_forwarder.go b/plugins/forwarder/grpc/nativejvm/sync_forwarder.go
index 68b3d86..6a1aa24 100644
--- a/plugins/forwarder/grpc/nativejvm/sync_forwarder.go
+++ b/plugins/forwarder/grpc/nativejvm/sync_forwarder.go
@@ -31,7 +31,10 @@ import (
"github.com/apache/skywalking-satellite/internal/satellite/event"
)
-const Name = "nativejvm-grpc-forwarder"
+const (
+ Name = "nativejvm-grpc-forwarder"
+ ShowName = "Native JVM GRPC Forwarder"
+)
type Forwarder struct {
config.CommonFields
@@ -43,6 +46,10 @@ func (f *Forwarder) Name() string {
return Name
}
+func (f *Forwarder) ShowName() string {
+ return ShowName
+}
+
func (f *Forwarder) Description() string {
return "This is a synchronization grpc forwarder with the SkyWalking native Configuration Discovery Service protocol."
}
diff --git a/plugins/forwarder/grpc/nativelog/forwarder.go b/plugins/forwarder/grpc/nativelog/forwarder.go
index 0faed35..8343f20 100644
--- a/plugins/forwarder/grpc/nativelog/forwarder.go
+++ b/plugins/forwarder/grpc/nativelog/forwarder.go
@@ -33,7 +33,10 @@ import (
"github.com/apache/skywalking-satellite/internal/satellite/event"
)
-const Name = "nativelog-grpc-forwarder"
+const (
+ Name = "nativelog-grpc-forwarder"
+ ShowName = "Native Log GRPC Forwarder"
+)
type Forwarder struct {
config.CommonFields
@@ -45,6 +48,10 @@ func (f *Forwarder) Name() string {
return Name
}
+func (f *Forwarder) ShowName() string {
+ return ShowName
+}
+
func (f *Forwarder) Description() string {
return "This is a synchronization grpc forwarder with the SkyWalking native log protocol."
}
diff --git a/plugins/forwarder/grpc/nativemanagement/forwarder.go b/plugins/forwarder/grpc/nativemanagement/forwarder.go
index 52409ad..217a11f 100644
--- a/plugins/forwarder/grpc/nativemanagement/forwarder.go
+++ b/plugins/forwarder/grpc/nativemanagement/forwarder.go
@@ -31,7 +31,10 @@ import (
v1 "skywalking.apache.org/repo/goapi/satellite/data/v1"
)
-const Name = "nativemanagement-grpc-forwarder"
+const (
+ Name = "nativemanagement-grpc-forwarder"
+ ShowName = "Native Management GRPC Forwarder"
+)
// empty struct for set
type void struct{}
@@ -48,6 +51,10 @@ func (f *Forwarder) Name() string {
return Name
}
+func (f *Forwarder) ShowName() string {
+ return ShowName
+}
+
func (f *Forwarder) Description() string {
return "This is a synchronization grpc forwarder with the SkyWalking native management protocol."
}
diff --git a/plugins/forwarder/grpc/nativemeter/forwarder.go b/plugins/forwarder/grpc/nativemeter/forwarder.go
index bd816ba..f3c172d 100644
--- a/plugins/forwarder/grpc/nativemeter/forwarder.go
+++ b/plugins/forwarder/grpc/nativemeter/forwarder.go
@@ -33,7 +33,10 @@ import (
"github.com/apache/skywalking-satellite/internal/satellite/event"
)
-const Name = "nativemeter-grpc-forwarder"
+const (
+ Name = "nativemeter-grpc-forwarder"
+ ShowName = "Native Meter GRPC Forwarder"
+)
type Forwarder struct {
config.CommonFields
@@ -44,6 +47,10 @@ func (f *Forwarder) Name() string {
return Name
}
+func (f *Forwarder) ShowName() string {
+ return ShowName
+}
+
func (f *Forwarder) Description() string {
return "This is a synchronization meter grpc forwarder with the SkyWalking meter protocol."
}
diff --git a/plugins/forwarder/grpc/nativeprofile/forwarder.go b/plugins/forwarder/grpc/nativeprofile/forwarder.go
index 7365bd8..f30fe6f 100644
--- a/plugins/forwarder/grpc/nativeprofile/forwarder.go
+++ b/plugins/forwarder/grpc/nativeprofile/forwarder.go
@@ -33,7 +33,10 @@ import (
"github.com/apache/skywalking-satellite/internal/satellite/event"
)
-const Name = "nativeprofile-grpc-forwarder"
+const (
+ Name = "nativeprofile-grpc-forwarder"
+ ShowName = "Native Profile GRPC Forwarder"
+)
type Forwarder struct {
config.CommonFields
@@ -45,6 +48,10 @@ func (f *Forwarder) Name() string {
return Name
}
+func (f *Forwarder) ShowName() string {
+ return ShowName
+}
+
func (f *Forwarder) Description() string {
return "This is a synchronization grpc forwarder with the SkyWalking native log protocol."
}
diff --git a/plugins/forwarder/grpc/nativetracing/forwarder.go b/plugins/forwarder/grpc/nativetracing/forwarder.go
index 41f7382..9087a3c 100644
--- a/plugins/forwarder/grpc/nativetracing/forwarder.go
+++ b/plugins/forwarder/grpc/nativetracing/forwarder.go
@@ -33,7 +33,10 @@ import (
v1 "skywalking.apache.org/repo/goapi/satellite/data/v1"
)
-const Name = "nativetracing-grpc-forwarder"
+const (
+ Name = "nativetracing-grpc-forwarder"
+ ShowName = "Native Tracing GRPC Forwarder"
+)
type Forwarder struct {
config.CommonFields
@@ -45,6 +48,10 @@ func (f *Forwarder) Name() string {
return Name
}
+func (f *Forwarder) ShowName() string {
+ return ShowName
+}
+
func (f *Forwarder) Description() string {
return "This is a synchronization grpc forwarder with the SkyWalking native tracing protocol."
}
diff --git a/plugins/forwarder/kafka/nativelog/forwarder.go b/plugins/forwarder/kafka/nativelog/forwarder.go
index de5b477..7e652e2 100644
--- a/plugins/forwarder/kafka/nativelog/forwarder.go
+++ b/plugins/forwarder/kafka/nativelog/forwarder.go
@@ -32,7 +32,10 @@ import (
v1 "skywalking.apache.org/repo/goapi/satellite/data/v1"
)
-const Name = "nativelog-kafka-forwarder"
+const (
+ Name = "nativelog-kafka-forwarder"
+ ShowName = "Native Log Kafka Forwarder"
+)
type Forwarder struct {
config.CommonFields
@@ -44,6 +47,10 @@ func (f *Forwarder) Name() string {
return Name
}
+func (f *Forwarder) ShowName() string {
+ return ShowName
+}
+
func (f *Forwarder) Description() string {
return "This is a synchronization Kafka forwarder with the SkyWalking native log protocol."
}
diff --git a/plugins/queue/api/queue.go b/plugins/queue/api/queue.go
index 8f2f1ee..a761423 100644
--- a/plugins/queue/api/queue.go
+++ b/plugins/queue/api/queue.go
@@ -43,13 +43,22 @@ type Queue interface {
Close() error
// Ack the lastOffset
- Ack(lastOffset event.Offset)
+ Ack(lastOffset *event.Offset)
+
+ // TotalSize total capacity of queue
+ TotalSize() int64
+
+ // UsedCount used count of queue
+ UsedCount() int64
+
+ // IsFull the queue is full
+ IsFull() bool
}
// SequenceEvent is a wrapper to pass the event and the offset.
type SequenceEvent struct {
Event *v1.SniffData
- Offset event.Offset
+ Offset *event.Offset
}
// GetQueue an initialized filter plugin.
diff --git a/plugins/queue/memory/queue.go b/plugins/queue/memory/queue.go
index bf89d78..629df83 100644
--- a/plugins/queue/memory/queue.go
+++ b/plugins/queue/memory/queue.go
@@ -29,9 +29,14 @@ import (
)
const (
- Name = "memory-queue"
+ Name = "memory-queue"
+ ShowName = "Memory Queue"
)
+var DefaultOffset = &event.Offset{
+ Position: "",
+}
+
type Queue struct {
config.CommonFields
// config
@@ -45,6 +50,10 @@ func (q *Queue) Name() string {
return Name
}
+func (q *Queue) ShowName() string {
+ return ShowName
+}
+
func (q *Queue) Description() string {
return "This is a memory queue to buffer the input event."
}
@@ -77,7 +86,7 @@ func (q *Queue) Dequeue() (*api.SequenceEvent, error) {
}
return &api.SequenceEvent{
Event: element.(*v1.SniffData),
- Offset: "no_offset_in_memory_queue",
+ Offset: DefaultOffset,
}, nil
}
@@ -85,5 +94,17 @@ func (q *Queue) Close() error {
return nil
}
-func (q *Queue) Ack(_ event.Offset) {
+func (q *Queue) Ack(_ *event.Offset) {
+}
+
+func (q *Queue) TotalSize() int64 {
+ return int64(q.EventBufferSize)
+}
+
+func (q *Queue) UsedCount() int64 {
+ return int64(q.buffer.GetLen())
+}
+
+func (q *Queue) IsFull() bool {
+ return q.UsedCount() >= q.TotalSize()
}
diff --git a/plugins/queue/mmap/queue.go b/plugins/queue/mmap/queue.go
index d7b0c2e..09ce199 100644
--- a/plugins/queue/mmap/queue.go
+++ b/plugins/queue/mmap/queue.go
@@ -15,7 +15,6 @@
// specific language governing permissions and limitations
// under the License.
-//go:build !windows
// +build !windows
package mmap
@@ -49,6 +48,7 @@ const (
data4KB = 131072
minimumSegments = 4
Name = "mmap-queue"
+ ShowName = "Mmap Queue"
)
// Queue is a memory mapped queue to store the input data.
@@ -73,6 +73,7 @@ type Queue struct {
markReadChannel chan int64 // Transfer the read segmentID to do ummap operation.
ready bool // The status of the queue.
locker []int32 // locker
+ usedSize int64 // The used size of queue
// control components
ctx context.Context // Parent ctx
@@ -84,6 +85,10 @@ func (q *Queue) Name() string {
return Name
}
+func (q *Queue) ShowName() string {
+ return ShowName
+}
+
func (q *Queue) Description() string {
return "This is a memory mapped queue to provide the persistent storage for the input event." +
" Please note that this plugin does not support Windows platform."
@@ -208,7 +213,7 @@ func (q *Queue) Close() error {
return nil
}
-func (q *Queue) Ack(lastOffset event.Offset) {
+func (q *Queue) Ack(lastOffset *event.Offset) {
if !q.ready {
log.Logger.WithFields(logrus.Fields{
"pipe": q.CommonFields.PipeName,
@@ -269,8 +274,7 @@ func (q *Queue) isEmpty() bool {
return rid == wid && roffset == woffset
}
-// isEmpty returns the capacity status
-func (q *Queue) isFull() bool {
+func (q *Queue) IsFull() bool {
rid, _ := q.meta.GetReadingOffset()
wid, _ := q.meta.GetWritingOffset()
// ensure enough spaces to promise data stability.
@@ -279,13 +283,13 @@ func (q *Queue) isFull() bool {
}
// encode the meta to the offset
-func (q *Queue) encodeOffset(id, offset int64) event.Offset {
- return event.Offset(strconv.FormatInt(id, 10) + "-" + strconv.FormatInt(offset, 10))
+func (q *Queue) encodeOffset(id, offset int64) *event.Offset {
+ return &event.Offset{Position: strconv.FormatInt(id, 10) + "-" + strconv.FormatInt(offset, 10)}
}
// decode the offset to the meta of the mmap queue.
-func (q *Queue) decodeOffset(val event.Offset) (id, offset int64, err error) {
- arr := strings.Split(string(val), "-")
+func (q *Queue) decodeOffset(val *event.Offset) (id, offset int64, err error) {
+ arr := strings.Split(val.Position, "-")
if len(arr) == 2 {
id, err := strconv.ParseInt(arr[0], 10, 64)
if err != nil {
@@ -299,3 +303,11 @@ func (q *Queue) decodeOffset(val event.Offset) (id, offset int64, err error) {
}
return 0, 0, fmt.Errorf("the input offset string is illegal: %s", val)
}
+
+func (q *Queue) TotalSize() int64 {
+ return int64(q.QueueCapacitySegments * q.SegmentSize)
+}
+
+func (q *Queue) UsedCount() int64 {
+ return q.usedSize
+}
diff --git a/plugins/queue/mmap/queue_operation.go b/plugins/queue/mmap/queue_operation.go
index 830399b..0b07ef0 100644
--- a/plugins/queue/mmap/queue_operation.go
+++ b/plugins/queue/mmap/queue_operation.go
@@ -25,6 +25,8 @@
package mmap
import (
+ "sync/atomic"
+
"github.com/apache/skywalking-satellite/plugins/queue/api"
)
@@ -43,11 +45,12 @@ const uInt64Size = 8
// enqueue writes the data into the file system. It first writes the length of the data,
// then the data itself. It means the whole data may not exist in the one segments.
func (q *Queue) enqueue(bytes []byte) error {
- if q.isFull() {
+ if q.IsFull() {
return api.ErrFull
}
id, offset := q.meta.GetWritingOffset()
- id, offset, err := q.writeLength(len(bytes), id, offset)
+ byteSize := len(bytes)
+ id, offset, err := q.writeLength(byteSize, id, offset)
if err != nil {
return err
}
@@ -61,6 +64,7 @@ func (q *Queue) enqueue(bytes []byte) error {
q.flushChannel <- struct{}{}
q.unflushedNum = 0
}
+ atomic.AddInt64(&q.usedSize, int64(byteSize))
return nil
}
@@ -83,6 +87,7 @@ func (q *Queue) dequeue() (data []byte, rid, roffset int64, err error) {
if id != preID {
q.markReadChannel <- preID
}
+ atomic.AddInt64(&q.usedSize, int64(-len(bytes)))
return bytes, id, offset, nil
}
diff --git a/plugins/queue/mmap/segment/segment.go b/plugins/queue/mmap/segment/segment.go
index 714c727..720769a 100644
--- a/plugins/queue/mmap/segment/segment.go
+++ b/plugins/queue/mmap/segment/segment.go
@@ -14,7 +14,7 @@
// KIND, either express or implied. See the License for the
// specific language governing permissions and limitations
// under the License.
-//go:build !windows
+
// +build !windows
package segment
diff --git a/plugins/queue/none/queue.go b/plugins/queue/none/queue.go
index 3c80df4..8e66aad 100644
--- a/plugins/queue/none/queue.go
+++ b/plugins/queue/none/queue.go
@@ -26,7 +26,8 @@ import (
)
const (
- Name = "none-queue"
+ Name = "none-queue"
+ ShowName = "None Queue"
)
type Queue struct {
@@ -37,6 +38,10 @@ func (q *Queue) Name() string {
return Name
}
+func (q *Queue) ShowName() string {
+ return ShowName
+}
+
func (q *Queue) Description() string {
return "This is an empty queue for direct connection protocols, such as SkyWalking native configuration discovery service protocol."
}
@@ -61,5 +66,17 @@ func (q *Queue) Close() error {
return nil
}
-func (q *Queue) Ack(_ event.Offset) {
+func (q *Queue) Ack(_ *event.Offset) {
+}
+
+func (q *Queue) TotalSize() int64 {
+ return 0
+}
+
+func (q *Queue) UsedCount() int64 {
+ return 0
+}
+
+func (q *Queue) IsFull() bool {
+ return false
}
diff --git a/plugins/queue/partition/partitioned_queue.go b/plugins/queue/partition/partitioned_queue.go
new file mode 100644
index 0000000..b8b01c0
--- /dev/null
+++ b/plugins/queue/partition/partitioned_queue.go
@@ -0,0 +1,169 @@
+// 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 partition
+
+import (
+ "fmt"
+ "reflect"
+ "strconv"
+ "sync/atomic"
+
+ "github.com/apache/skywalking-satellite/internal/pkg/config"
+ "github.com/apache/skywalking-satellite/internal/pkg/plugin"
+ "github.com/apache/skywalking-satellite/internal/satellite/event"
+ "github.com/apache/skywalking-satellite/internal/satellite/telemetry"
+ "github.com/apache/skywalking-satellite/plugins/queue/api"
+
+ v1 "skywalking.apache.org/repo/goapi/satellite/data/v1"
+)
+
+type PartitionedQueue struct {
+ config.CommonFields
+ Partition int `mapstructure:"partition"` // The total partition count.
+ config plugin.Config
+
+ subQueues []api.Queue
+ loadBalancerIndex int32
+ totalQueueSize int64
+}
+
+func NewPartitionQueue(c plugin.Config) *PartitionedQueue {
+ queue := &PartitionedQueue{config: c}
+ plugin.Initializing(queue, c)
+ return queue
+}
+
+func (p *PartitionedQueue) Initialize() error {
+ queues := make([]api.Queue, p.Partition)
+ pipeName := p.PipeName
+ for partition := 0; partition < p.Partition; partition++ {
+ p.config["pipe_name"] = fmt.Sprintf("%s-%d", p.config["pipe_name"], partition)
+ queue := plugin.Get(reflect.TypeOf((*api.Queue)(nil)).Elem(), p.config).(api.Queue)
+ if err := queue.Initialize(); err != nil {
+ return err
+ }
+ queues[partition] = queue
+ p.totalQueueSize += queue.TotalSize()
+ }
+ p.subQueues = queues
+ p.registerQueueTelemetry(pipeName)
+ return nil
+}
+
+func (p *PartitionedQueue) DefaultConfig() string {
+ return `
+# The partition count of queue.
+partition: 1
+`
+}
+
+func (p *PartitionedQueue) Enqueue(e *v1.SniffData) error {
+ partition, err := p.findPartition(e)
+ if err != nil {
+ return err
+ }
+ return p.subQueues[partition].Enqueue(e)
+}
+
+func (p *PartitionedQueue) Dequeue(partition int) (*api.SequenceEvent, error) {
+ dequeue, err := p.subQueues[partition].Dequeue()
+ if err == nil {
+ dequeue.Offset.Partition = partition
+ }
+ return dequeue, err
+}
+
+func (p *PartitionedQueue) TotalPartitionCount() int {
+ return len(p.subQueues)
+}
+
+func (p *PartitionedQueue) Name() string {
+ return "partition-queue"
+}
+
+func (p *PartitionedQueue) ShowName() string {
+ return "Partition Queue"
+}
+
+func (p *PartitionedQueue) Description() string {
+ return "The partition queue to management all the sub queues."
+}
+
+func (p *PartitionedQueue) Close() error {
+ for _, q := range p.subQueues {
+ if err := q.Close(); err != nil {
+ return err
+ }
+ }
+ return nil
+}
+
+func (p *PartitionedQueue) Ack(lastOffset *event.Offset) {
+ p.subQueues[lastOffset.Partition].Ack(lastOffset)
+}
+
+func (p *PartitionedQueue) findPartition(_ *v1.SniffData) (int, error) {
+ if p.Partition == 1 {
+ return 0, nil
+ }
+
+ // increment
+ var partition int
+ for {
+ result := atomic.AddInt32(&p.loadBalancerIndex, 1)
+ partition = int(result)
+
+ if partition < p.Partition {
+ break
+ } else if atomic.CompareAndSwapInt32(&p.loadBalancerIndex, result, 0) {
+ partition = 0
+ break
+ }
+ }
+
+ // check partition is full
+ if !p.subQueues[partition].IsFull() {
+ return partition, nil
+ }
+ for addition := 1; addition < p.Partition; addition++ {
+ checkPartition := partition + addition
+ if checkPartition >= p.Partition {
+ checkPartition -= p.Partition
+ }
+ if !p.subQueues[checkPartition].IsFull() {
+ return checkPartition, nil
+ }
+ }
+ return 0, fmt.Errorf("the queue is full")
+}
+
+func (p *PartitionedQueue) registerQueueTelemetry(pipeline string) {
+ telemetry.NewGauge("pipeline_queue_total_capacity", "The total capacity of pipeline", func() float64 {
+ return float64(p.totalQueueSize)
+ }, "pipeline", pipeline)
+
+ for partition := 0; partition < len(p.subQueues); partition++ {
+ pn := partition
+ telemetry.NewGauge("pipeline_queue_partition_size",
+ "The current count of elements in the queue partition",
+ func() float64 {
+ return float64(p.subQueues[pn].UsedCount())
+ },
+ "pipeline", pipeline, "partition", strconv.Itoa(pn))
+ }
+}
diff --git a/plugins/queue/partition/partitioned_queue_test.go b/plugins/queue/partition/partitioned_queue_test.go
new file mode 100644
index 0000000..8593721
--- /dev/null
+++ b/plugins/queue/partition/partitioned_queue_test.go
@@ -0,0 +1,145 @@
+// 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 partition
+
+import (
+ "fmt"
+ "reflect"
+ "strconv"
+ "testing"
+
+ "github.com/apache/skywalking-satellite/internal/pkg/log"
+ "github.com/apache/skywalking-satellite/internal/pkg/plugin"
+ "github.com/apache/skywalking-satellite/internal/satellite/telemetry"
+ "github.com/apache/skywalking-satellite/plugins/queue/api"
+ "github.com/apache/skywalking-satellite/plugins/queue/memory"
+
+ v1 "skywalking.apache.org/repo/goapi/satellite/data/v1"
+)
+
+func init() {
+ log.Init(&log.LoggerConfig{})
+ telemetry.Init(&telemetry.Config{})
+}
+
+func initPartitionQueue(queueName string, cfg plugin.Config) (*PartitionedQueue, error) {
+ plugin.RegisterPluginCategory(reflect.TypeOf((*api.Queue)(nil)).Elem())
+ plugin.RegisterPlugin(&memory.Queue{})
+ var config plugin.Config = map[string]interface{}{
+ plugin.NameField: queueName,
+ }
+ for k, v := range cfg {
+ config[k] = v
+ }
+ q := NewPartitionQueue(config)
+ if q == nil {
+ return nil, fmt.Errorf("cannot get a partition queue from the registry")
+ }
+ if err := q.Initialize(); err != nil {
+ return nil, fmt.Errorf("queue cannot initialize: %v", err)
+ }
+ return q, nil
+}
+
+func initData(count int) []*v1.SniffData {
+ result := make([]*v1.SniffData, count)
+ for inx := range result {
+ result[inx] = &v1.SniffData{Name: strconv.Itoa(inx)}
+ }
+ return result
+}
+
+func TestPartitionQueue(t *testing.T) {
+ tests := []struct {
+ name string
+ bufferSize int
+ partitionCount int
+ initDataCount int
+ loadBalancerIndex int
+ testCount int
+ enqueueError bool
+ }{
+ {
+ name: "normal simple queue",
+ bufferSize: 10,
+ partitionCount: 1,
+ initDataCount: 2,
+ loadBalancerIndex: 2,
+ testCount: 6,
+ enqueueError: false,
+ },
+ {
+ name: "normal multiple queue",
+ bufferSize: 10,
+ partitionCount: 5,
+ initDataCount: 10,
+ loadBalancerIndex: 0,
+ testCount: 10,
+ enqueueError: false,
+ },
+ {
+ name: "single partition full",
+ bufferSize: 2,
+ partitionCount: 5,
+ initDataCount: 8,
+ loadBalancerIndex: 2,
+ testCount: 2,
+ enqueueError: false,
+ },
+ {
+ name: "all partition full",
+ bufferSize: 2,
+ partitionCount: 5,
+ initDataCount: 10,
+ loadBalancerIndex: 1,
+ testCount: 1,
+ enqueueError: true,
+ },
+ }
+
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ queue, err := initPartitionQueue("memory-queue", map[string]interface{}{
+ "event_buffer_size": tt.bufferSize,
+ "partition": tt.partitionCount,
+ })
+ if err != nil {
+ t.Fatal(err)
+ }
+
+ initDates := initData(tt.initDataCount)
+ for _, d := range initDates {
+ if err2 := queue.Enqueue(d); err2 != nil {
+ t.Fatal(err2)
+ }
+ }
+
+ queue.loadBalancerIndex = int32(tt.loadBalancerIndex)
+
+ testDates := initData(tt.testCount)
+ for _, d := range testDates {
+ err = queue.Enqueue(d)
+ }
+ if tt.enqueueError && err == nil {
+ t.Fatalf("should contains enqueue error")
+ } else if !tt.enqueueError && err != nil {
+ t.Fatalf("should not contain enqueue error, error: %v", err)
+ }
+ })
+ }
+}
diff --git a/plugins/receiver/grpc/envoyalsv2/receiver.go b/plugins/receiver/grpc/envoyalsv2/receiver.go
index 3fad9f2..c3bb019 100644
--- a/plugins/receiver/grpc/envoyalsv2/receiver.go
+++ b/plugins/receiver/grpc/envoyalsv2/receiver.go
@@ -29,7 +29,10 @@ import (
v1 "skywalking.apache.org/repo/goapi/satellite/data/v1"
)
-const Name = "grpc-envoy-als-v2-receiver"
+const (
+ Name = "grpc-envoy-als-v2-receiver"
+ ShowName = "GRPC Envoy ALS v2 Receiver"
+)
type Receiver struct {
config.CommonFields
@@ -43,6 +46,10 @@ func (r *Receiver) Name() string {
return Name
}
+func (r *Receiver) ShowName() string {
+ return ShowName
+}
+
func (r *Receiver) Description() string {
return "This is a receiver for Envoy ALS format, " +
"which is defined at https://github.com/envoyproxy/envoy/blob/v1.17.4/api/envoy/service/accesslog/v2/als.proto."
diff --git a/plugins/receiver/grpc/envoyalsv3/als_service.go b/plugins/receiver/grpc/envoyalsv3/als_service.go
index 127774e..ef061c1 100644
--- a/plugins/receiver/grpc/envoyalsv3/als_service.go
+++ b/plugins/receiver/grpc/envoyalsv3/als_service.go
@@ -23,6 +23,7 @@ import (
"time"
"github.com/apache/skywalking-satellite/internal/satellite/module/buffer"
+ "github.com/apache/skywalking-satellite/internal/satellite/telemetry"
v3 "skywalking.apache.org/repo/goapi/proto/envoy/service/accesslog/v3"
v1 "skywalking.apache.org/repo/goapi/satellite/data/v1"
@@ -34,6 +35,22 @@ type AlsService struct {
receiveChannel chan *v1.SniffData
limiterConfig buffer.LimiterConfig
v3.UnimplementedAccessLogServiceServer
+
+ messageCount *telemetry.Counter
+ streamingCount *telemetry.Counter
+ streamingFailedCount *telemetry.Counter
+ streamingToEventCount *telemetry.Counter
+}
+
+func (m *AlsService) init() {
+ m.messageCount = telemetry.NewCounter("als_message_receive_count",
+ "Total count of the receive message in the ALS Receiver.")
+ m.streamingCount = telemetry.NewCounter("als_streaming_receive_count",
+ "Total count of the receive stream in the ALS Receiver.")
+ m.streamingFailedCount = telemetry.NewCounter("als_streaming_receive_failed_count",
+ "Total count of the failed receive message in the ALS Receiver.")
+ m.streamingToEventCount = telemetry.NewCounter("als_streaming_to_event_count",
+ "Total count of the packaged the ALS streaming in the ALS Receiver.")
}
func (m *AlsService) StreamAccessLogs(stream v3.AccessLogService_StreamAccessLogsServer) error {
@@ -68,6 +85,7 @@ func (m *AlsService) StreamAccessLogs(stream v3.AccessLogService_StreamAccessLog
},
},
}
+ m.streamingToEventCount.Inc()
m.receiveChannel <- d
})
@@ -75,12 +93,19 @@ func (m *AlsService) StreamAccessLogs(stream v3.AccessLogService_StreamAccessLog
for {
item, err := stream.Recv()
if err != nil {
+ m.streamingFailedCount.Inc()
err1 = err
break
}
if item.Identifier != nil {
identity = item.Identifier
}
+ m.streamingCount.Inc()
+ if item.GetHttpLogs() != nil {
+ m.messageCount.Add(float64(len(item.GetHttpLogs().LogEntry)))
+ } else if item.GetTcpLogs() != nil {
+ m.messageCount.Add(float64(len(item.GetTcpLogs().LogEntry)))
+ }
messages <- item
limiter.Check()
}
diff --git a/plugins/receiver/grpc/envoyalsv3/receiver.go b/plugins/receiver/grpc/envoyalsv3/receiver.go
index b63d163..3f0f23c 100644
--- a/plugins/receiver/grpc/envoyalsv3/receiver.go
+++ b/plugins/receiver/grpc/envoyalsv3/receiver.go
@@ -29,7 +29,10 @@ import (
v1 "skywalking.apache.org/repo/goapi/satellite/data/v1"
)
-const Name = "grpc-envoy-als-v3-receiver"
+const (
+ Name = "grpc-envoy-als-v3-receiver"
+ ShowName = "GRPC Envoy ALS v3 Receiver"
+)
type Receiver struct {
config.CommonFields
@@ -43,6 +46,10 @@ func (r *Receiver) Name() string {
return Name
}
+func (r *Receiver) ShowName() string {
+ return ShowName
+}
+
func (r *Receiver) Description() string {
return "This is a receiver for Envoy ALS format, " +
"which is defined at https://github.com/envoyproxy/envoy/blob/" +
@@ -61,6 +68,7 @@ limit_count: 500
func (r *Receiver) RegisterHandler(server interface{}) {
r.CommonGRPCReceiverFields = *grpc.InitCommonGRPCReceiverFields(server)
r.service = &AlsService{receiveChannel: r.OutputChannel, limiterConfig: r.LimitConfig}
+ r.service.init()
v3.RegisterAccessLogServiceServer(r.Server, r.service)
}
diff --git a/plugins/receiver/grpc/envoymetricsv2/receiver.go b/plugins/receiver/grpc/envoymetricsv2/receiver.go
index 82b5560..304b54b 100644
--- a/plugins/receiver/grpc/envoymetricsv2/receiver.go
+++ b/plugins/receiver/grpc/envoymetricsv2/receiver.go
@@ -29,7 +29,10 @@ import (
v1 "skywalking.apache.org/repo/goapi/satellite/data/v1"
)
-const Name = "grpc-envoy-metrics-v2-receiver"
+const (
+ Name = "grpc-envoy-metrics-v2-receiver"
+ ShowName = "GRPC Envoy Metrics v2 Receiver"
+)
type Receiver struct {
config.CommonFields
@@ -43,6 +46,10 @@ func (r *Receiver) Name() string {
return Name
}
+func (r *Receiver) ShowName() string {
+ return ShowName
+}
+
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."
diff --git a/plugins/receiver/grpc/envoymetricsv3/receiver.go b/plugins/receiver/grpc/envoymetricsv3/receiver.go
index b3da8b6..00cef67 100644
--- a/plugins/receiver/grpc/envoymetricsv3/receiver.go
+++ b/plugins/receiver/grpc/envoymetricsv3/receiver.go
@@ -29,7 +29,10 @@ import (
v1 "skywalking.apache.org/repo/goapi/satellite/data/v1"
)
-const Name = "grpc-envoy-metrics-v3-receiver"
+const (
+ Name = "grpc-envoy-metrics-v3-receiver"
+ ShowName = "GRPC Envoy Metrics v3 Receiver"
+)
type Receiver struct {
config.CommonFields
@@ -43,6 +46,10 @@ func (r *Receiver) Name() string {
return Name
}
+func (r *Receiver) ShowName() string {
+ return ShowName
+}
+
func (r *Receiver) Description() string {
return "This is a receiver for Envoy Metrics format, " +
"which is defined at https://github.com/envoyproxy/envoy/blob/" +
diff --git a/plugins/receiver/grpc/nativecds/receiver.go b/plugins/receiver/grpc/nativecds/receiver.go
index dadc9f4..d2cadf4 100644
--- a/plugins/receiver/grpc/nativecds/receiver.go
+++ b/plugins/receiver/grpc/nativecds/receiver.go
@@ -28,7 +28,10 @@ import (
v1 "skywalking.apache.org/repo/goapi/satellite/data/v1"
)
-const Name = "grpc-nativecds-receiver"
+const (
+ Name = "grpc-nativecds-receiver"
+ ShowName = "GRPC Native CDS Receiver"
+)
type Receiver struct {
config.CommonFields
@@ -40,6 +43,10 @@ func (r *Receiver) Name() string {
return Name
}
+func (r *Receiver) ShowName() string {
+ return ShowName
+}
+
func (r *Receiver) Description() string {
return "This is a receiver for SkyWalking native Configuration Discovery Service format, " +
"which is defined at https://github.com/apache/skywalking-data-collect-protocol/blob/master/language-agent/ConfigurationDiscoveryService.proto."
diff --git a/plugins/receiver/grpc/nativeevent/receiver.go b/plugins/receiver/grpc/nativeevent/receiver.go
index efb4572..4b591ed 100644
--- a/plugins/receiver/grpc/nativeevent/receiver.go
+++ b/plugins/receiver/grpc/nativeevent/receiver.go
@@ -28,7 +28,10 @@ import (
v1 "skywalking.apache.org/repo/goapi/satellite/data/v1"
)
-const Name = "grpc-nativeevent-receiver"
+const (
+ Name = "grpc-nativeevent-receiver"
+ ShowName = "GRPC Native Event Receiver"
+)
type Receiver struct {
config.CommonFields
@@ -40,6 +43,10 @@ func (r *Receiver) Name() string {
return Name
}
+func (r *Receiver) ShowName() string {
+ return ShowName
+}
+
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."
diff --git a/plugins/receiver/grpc/nativejvm/receiver.go b/plugins/receiver/grpc/nativejvm/receiver.go
index 8d85234..5028fbe 100644
--- a/plugins/receiver/grpc/nativejvm/receiver.go
+++ b/plugins/receiver/grpc/nativejvm/receiver.go
@@ -28,7 +28,10 @@ import (
grpcreceiver "github.com/apache/skywalking-satellite/plugins/receiver/grpc"
)
-const Name = "grpc-nativejvm-receiver"
+const (
+ Name = "grpc-nativejvm-receiver"
+ ShowName = "GRPC Native JVM Receiver"
+)
type Receiver struct {
config.CommonFields
@@ -40,6 +43,10 @@ func (r *Receiver) Name() string {
return Name
}
+func (r *Receiver) ShowName() string {
+ return ShowName
+}
+
func (r *Receiver) Description() string {
return "This is a receiver for SkyWalking native jvm format, " +
"which is defined at https://github.com/apache/skywalking-data-collect-protocol/blob/master/language-agent/JVMMetric.proto."
diff --git a/plugins/receiver/grpc/nativelog/receiver.go b/plugins/receiver/grpc/nativelog/receiver.go
index b8f32e3..d293968 100644
--- a/plugins/receiver/grpc/nativelog/receiver.go
+++ b/plugins/receiver/grpc/nativelog/receiver.go
@@ -28,7 +28,10 @@ import (
grpcreceiver "github.com/apache/skywalking-satellite/plugins/receiver/grpc"
)
-const Name = "grpc-nativelog-receiver"
+const (
+ Name = "grpc-nativelog-receiver"
+ ShowName = "GRPC Native Log Receiver"
+)
type Receiver struct {
config.CommonFields
@@ -40,6 +43,10 @@ func (r *Receiver) Name() string {
return Name
}
+func (r *Receiver) ShowName() string {
+ return ShowName
+}
+
func (r *Receiver) Description() string {
return "This is a receiver for SkyWalking native logging format, " +
"which is defined at https://github.com/apache/skywalking-data-collect-protocol/blob/master/logging/Logging.proto."
diff --git a/plugins/receiver/grpc/nativemanagement/receiver.go b/plugins/receiver/grpc/nativemanagement/receiver.go
index 84372ae..edb47ce 100644
--- a/plugins/receiver/grpc/nativemanagement/receiver.go
+++ b/plugins/receiver/grpc/nativemanagement/receiver.go
@@ -28,7 +28,10 @@ import (
v1 "skywalking.apache.org/repo/goapi/satellite/data/v1"
)
-const Name = "grpc-nativemanagement-receiver"
+const (
+ Name = "grpc-nativemanagement-receiver"
+ ShowName = "GRPC Native Management Receiver"
+)
type Receiver struct {
config.CommonFields
@@ -40,6 +43,10 @@ func (r *Receiver) Name() string {
return Name
}
+func (r *Receiver) ShowName() string {
+ return ShowName
+}
+
func (r *Receiver) Description() string {
return "This is a receiver for SkyWalking native management format, " +
"which is defined at https://github.com/apache/skywalking-data-collect-protocol/blob/master/management/Management.proto."
diff --git a/plugins/receiver/grpc/nativemeter/receiver.go b/plugins/receiver/grpc/nativemeter/receiver.go
index b6b623c..ba90356 100644
--- a/plugins/receiver/grpc/nativemeter/receiver.go
+++ b/plugins/receiver/grpc/nativemeter/receiver.go
@@ -28,7 +28,10 @@ import (
v1 "skywalking.apache.org/repo/goapi/satellite/data/v1"
)
-const Name = "grpc-nativemeter-receiver"
+const (
+ Name = "grpc-nativemeter-receiver"
+ ShowName = "GRPC Native Meter Receiver"
+)
type Receiver struct {
config.CommonFields
@@ -40,6 +43,10 @@ func (r *Receiver) Name() string {
return Name
}
+func (r *Receiver) ShowName() string {
+ return ShowName
+}
+
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."
diff --git a/plugins/receiver/grpc/nativeprofile/receiver.go b/plugins/receiver/grpc/nativeprofile/receiver.go
index 8bf44c2..33aabdc 100644
--- a/plugins/receiver/grpc/nativeprofile/receiver.go
+++ b/plugins/receiver/grpc/nativeprofile/receiver.go
@@ -28,7 +28,10 @@ import (
v1 "skywalking.apache.org/repo/goapi/satellite/data/v1"
)
-const Name = "grpc-nativeprofile-receiver"
+const (
+ Name = "grpc-nativeprofile-receiver"
+ ShowName = "GRPC Native Profile Receiver"
+)
type Receiver struct {
config.CommonFields
@@ -40,6 +43,10 @@ func (r *Receiver) Name() string {
return Name
}
+func (r *Receiver) ShowName() string {
+ return ShowName
+}
+
func (r *Receiver) Description() string {
return "This is a receiver for SkyWalking native profile format, " +
"which is defined at https://github.com/apache/skywalking-data-collect-protocol/blob/master/profile/Profile.proto."
diff --git a/plugins/receiver/grpc/nativetracing/receiver.go b/plugins/receiver/grpc/nativetracing/receiver.go
index 7373c92..c566733 100644
--- a/plugins/receiver/grpc/nativetracing/receiver.go
+++ b/plugins/receiver/grpc/nativetracing/receiver.go
@@ -28,7 +28,10 @@ import (
v1 "skywalking.apache.org/repo/goapi/satellite/data/v1"
)
-const Name = "grpc-nativetracing-receiver"
+const (
+ Name = "grpc-nativetracing-receiver"
+ ShowName = "GRPC Native Tracing Receiver"
+)
type Receiver struct {
config.CommonFields
@@ -40,6 +43,10 @@ func (r *Receiver) Name() string {
return Name
}
+func (r *Receiver) ShowName() string {
+ return ShowName
+}
+
func (r *Receiver) Description() string {
return "This is a receiver for SkyWalking native tracing format, " +
"which is defined at https://github.com/apache/skywalking-data-collect-protocol/blob/master/language-agent/Tracing.proto."
diff --git a/plugins/receiver/http/nativcelog/receiver.go b/plugins/receiver/http/nativcelog/receiver.go
index 82c599c..e30949a 100644
--- a/plugins/receiver/http/nativcelog/receiver.go
+++ b/plugins/receiver/http/nativcelog/receiver.go
@@ -40,6 +40,7 @@ import (
const (
Name = "http-nativelog-receiver"
+ ShowName = "HTTP Native Log Receiver"
eventName = "http-nativelog-event"
success = "success"
failing = "failing"
@@ -64,6 +65,10 @@ func (r *Receiver) Name() string {
return Name
}
+func (r *Receiver) ShowName() string {
+ return ShowName
+}
+
func (r *Receiver) Description() string {
return "This is a receiver for SkyWalking http logging format, " +
"which is defined at https://github.com/apache/skywalking-data-collect-protocol/blob/master/logging/Logging.proto."
diff --git a/plugins/server/grpc/server.go b/plugins/server/grpc/server.go
index 3351823..a3f9f17 100644
--- a/plugins/server/grpc/server.go
+++ b/plugins/server/grpc/server.go
@@ -27,7 +27,10 @@ import (
"github.com/apache/skywalking-satellite/internal/pkg/log"
)
-const Name = "grpc-server"
+const (
+ Name = "grpc-server"
+ ShowName = "GRPC Server"
+)
type Server struct {
config.CommonFields
@@ -46,6 +49,10 @@ func (s *Server) Name() string {
return Name
}
+func (s *Server) ShowName() string {
+ return ShowName
+}
+
func (s *Server) Description() string {
return "This is a sharing plugin, which would start a gRPC server."
}
diff --git a/plugins/server/http/server.go b/plugins/server/http/server.go
index 7c7810d..83e2850 100644
--- a/plugins/server/http/server.go
+++ b/plugins/server/http/server.go
@@ -24,7 +24,10 @@ import (
"github.com/apache/skywalking-satellite/internal/pkg/log"
)
-const Name = "http-server"
+const (
+ Name = "http-server"
+ ShowName = "HTTP Server"
+)
type Server struct {
config.CommonFields
@@ -36,6 +39,10 @@ func (s *Server) Name() string {
return Name
}
+func (s *Server) ShowName() string {
+ return ShowName
+}
+
func (s *Server) Description() string {
return "This is a sharing plugin, which would start a http server."
}
diff --git a/plugins/server/prometheus/prometheus.go b/plugins/server/prometheus/prometheus.go
index 8df5e77..b68d418 100644
--- a/plugins/server/prometheus/prometheus.go
+++ b/plugins/server/prometheus/prometheus.go
@@ -28,7 +28,10 @@ import (
"github.com/apache/skywalking-satellite/internal/satellite/telemetry"
)
-const Name = "prometheus-server"
+const (
+ Name = "prometheus-server"
+ ShowName = "Prometheus Server"
+)
type Server struct {
config.CommonFields
@@ -42,6 +45,10 @@ func (s *Server) Name() string {
return Name
}
+func (s *Server) ShowName() string {
+ return ShowName
+}
+
func (s *Server) Description() string {
return "This is a prometheus server to export the metrics in Satellite."
}