You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@skywalking.apache.org by li...@apache.org on 2025/06/16 15:38:24 UTC
(skywalking-go) branch main updated: Kafka Reporter for Enhanced Data Reporting (#224)
This is an automated email from the ASF dual-hosted git repository.
liuhan pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/skywalking-go.git
The following commit(s) were added to refs/heads/main by this push:
new cf6e915 Kafka Reporter for Enhanced Data Reporting (#224)
cf6e915 is described below
commit cf6e9156ddcad90617ad442fb354cecf5e6017f4
Author: zhujiajun <zh...@gmail.com>
AuthorDate: Mon Jun 16 23:38:19 2025 +0800
Kafka Reporter for Enhanced Data Reporting (#224)
---
.github/workflows/e2e.yaml | 64 ++++
.github/workflows/windows-plugin-test.yaml | 29 +-
.gitignore | 1 +
CHANGES.md | 1 +
Makefile | 2 +-
agent/reporter/imports.go | 6 +
docs/en/advanced-features/kafka-reporter.md | 148 +++++++++
docs/menu.yml | 2 +
go.work | 3 +
go.work.sum | 6 +-
plugins/core/go.mod | 1 +
plugins/core/metrics.go | 3 +
plugins/core/reporter/cds_manager.go | 96 ++++++
plugins/core/reporter/conn_manager.go | 218 +++++++++++++
plugins/core/reporter/grpc/grpc.go | 291 ++++-------------
plugins/core/reporter/grpc/grpc_opts_enhance.go | 83 -----
plugins/core/reporter/kafka/kafka.go | 352 +++++++++++++++++++++
plugins/core/reporter/kafka/kafka_opts_enhance.go | 85 +++++
plugins/core/reporter/transform.go | 151 +++++++++
plugins/segmentio-kafka/instrument.go | 2 +-
plugins/segmentio-kafka/writer_interceptor.go | 8 +
test/e2e/Makefile | 39 +++
test/e2e/base/consumer/Dockerfile | 40 +++
test/e2e/base/consumer/go.mod | 3 +
test/e2e/base/consumer/main.go | 133 ++++++++
test/e2e/base/docker-compose.yml | 56 ++++
test/e2e/base/provider/Dockerfile | 40 +++
test/e2e/base/provider/go.mod | 3 +
test/e2e/base/provider/main.go | 83 +++++
test/e2e/case/expected/logs-list.yml | 88 ++++++
test/e2e/case/expected/metrics-has-value.yml | 21 ++
.../case/expected/service-endpoint-consumer.yml | 19 ++
.../case/expected/service-endpoint-provider.yml | 19 ++
.../case/expected/service-instance-consumer.yml | 34 ++
.../case/expected/service-instance-provider.yml | 36 +++
test/e2e/case/expected/service.yml | 33 ++
test/e2e/case/expected/trace-info-detail.yml | 47 +++
test/e2e/case/expected/trace-users-detail.yml | 43 +++
test/e2e/case/expected/traces-list-consumer.yml | 29 ++
test/e2e/case/expected/traces-list-provider.yml | 29 ++
test/e2e/case/grpc/docker-compose.yml | 64 ++++
test/e2e/case/grpc/e2e.yaml | 88 ++++++
test/e2e/case/kafka/docker-compose.yml | 141 +++++++++
test/e2e/case/kafka/e2e.yaml | 88 ++++++
test/e2e/script/docker-compose/base-compose.yml | 61 ++++
test/e2e/script/env | 19 ++
test/e2e/script/prepare/setup-e2e/install-swctl.sh | 39 +++
test/e2e/script/prepare/setup-e2e/install-yq.sh | 30 ++
test/e2e/script/prepare/setup-e2e/install.sh | 35 ++
test/e2e/script/prepare/setup-oap/log4j2.xml | 33 ++
.../templates/windows-docker-compose.tpl | 2 +-
.../runner-helper/templates/windows-scenarios.tpl | 2 +
test/plugins/scenarios/goframe/plugin.yml | 2 +-
test/plugins/scenarios/gorm/plugin.yml | 2 +-
tools/go-agent/config/agent.default.yaml | 31 +-
tools/go-agent/config/loader.go | 28 +-
tools/go-agent/instrument/agentcore/instrument.go | 6 +-
tools/go-agent/instrument/consts/reporter.go | 4 +-
tools/go-agent/instrument/instrument.go | 2 +-
tools/go-agent/instrument/reporter/instrument.go | 192 ++++++++---
60 files changed, 2832 insertions(+), 384 deletions(-)
diff --git a/.github/workflows/e2e.yaml b/.github/workflows/e2e.yaml
new file mode 100644
index 0000000..7b956b0
--- /dev/null
+++ b/.github/workflows/e2e.yaml
@@ -0,0 +1,64 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+name: E2E
+
+on:
+ pull_request:
+
+jobs:
+ build:
+ name: Build SkyWalking Go agent
+ runs-on: ubuntu-latest
+ steps:
+ - name: Set up Go
+ uses: actions/setup-go@v2
+ with:
+ go-version: 1.19
+ - uses: actions/checkout@v2
+ with:
+ submodules: true
+ - name: Build SkyWalking Go agent
+ run: make -C test/e2e build
+ - name: Upload agent binary
+ uses: actions/upload-artifact@v4
+ with:
+ name: skywalking-go-agent
+ path: test/e2e/dist/
+ e2e:
+ name: Basic function + ${{ matrix.case.name }} reporter
+ runs-on: ubuntu-latest
+ needs: build
+ timeout-minutes: 90
+ strategy:
+ matrix:
+ case:
+ - name: gRPC
+ path: test/e2e/case/grpc/e2e.yaml
+ - name: Kafka
+ path: test/e2e/case/kafka/e2e.yaml
+ steps:
+ - uses: actions/checkout@v2
+ with:
+ submodules: true
+ - name: Download agent binary
+ uses: actions/download-artifact@v4
+ with:
+ name: skywalking-go-agent
+ path: test/e2e/dist/
+ - name: Run E2E Tests
+ uses: apache/skywalking-infra-e2e@7e4b5b68716fdb7b79b21fa8908f9db497e1b115
+ with:
+ e2e-file: ${{ matrix.case.path }}
diff --git a/.github/workflows/windows-plugin-test.yaml b/.github/workflows/windows-plugin-test.yaml
index e6b14cd..e4f3c92 100644
--- a/.github/workflows/windows-plugin-test.yaml
+++ b/.github/workflows/windows-plugin-test.yaml
@@ -38,12 +38,17 @@ jobs:
Enable-WindowsOptionalFeature -Online -FeatureName Microsoft-Windows-Subsystem-Linux
Enable-WindowsOptionalFeature -Online -FeatureName VirtualMachinePlatform
- Invoke-WebRequest -Uri https://wslstorestorage.blob.core.windows.net/wslblob/wsl_update_x64.msi -OutFile wsl_update.msi
- Start-Process msiexec.exe -Wait -ArgumentList '/i wsl_update.msi /quiet'
-
wsl --set-default-version 2
wsl --update
- wsl --install -d Ubuntu
+
+ $distros = (Invoke-RestMethod -Uri https://raw.githubusercontent.com/microsoft/WSL/refs/heads/master/distributions/DistributionInfo.json).Distributions
+ $distroDownloadUrl = $distros.Where({ $_.Name -eq 'Ubuntu' }).Amd64PackageUrl
+ Set-Location $env:GITHUB_WORKSPACE\..\..
+ $filename = "Ubuntu$([System.IO.Path]::GetExtension($distroDownloadUrl))"
+ Invoke-WebRequest -Uri $distroDownloadUrl -OutFile $filename
+ Expand-Archive -Path $filename -DestinationPath .\
+ $distroExe = (Get-ChildItem -Path . -Filter *.exe).Where({ $_.Name -notmatch 'splash|setup' }).FullName
+ & $distroExe install --root
@'
@echo off
@@ -56,19 +61,27 @@ jobs:
- name: WSL - Setup Docker
shell: wsl-run {0}
run: |
- # Add Docker's official GPG key:
- sudo apt-get update
+ for i in {1..3}; do
+ sudo rm -r /var/lib/apt/lists/*
+ sudo mkdir /var/lib/apt/lists/partial
+ sudo apt-get update -qq && break || echo "apt-get update failed, retrying ($i)…"
+ done
+
sudo apt-get install -y ca-certificates curl
sudo install -m 0755 -d /etc/apt/keyrings
sudo curl -fsSL https://download.docker.com/linux/ubuntu/gpg -o /etc/apt/keyrings/docker.asc
sudo chmod a+r /etc/apt/keyrings/docker.asc
- # Add the repository to Apt sources:
echo \
"deb [arch=$(dpkg --print-architecture) signed-by=/etc/apt/keyrings/docker.asc] https://download.docker.com/linux/ubuntu \
$(. /etc/os-release && echo "$VERSION_CODENAME") stable" | \
sudo tee /etc/apt/sources.list.d/docker.list > /dev/null
- sudo apt-get update
+
+ for i in {1..3}; do
+ sudo rm -r /var/lib/apt/lists/*
+ sudo mkdir /var/lib/apt/lists/partial
+ sudo apt-get update -qq && break || echo "apt-get update failed, retrying ($i)…"
+ done
sudo apt-get install -y docker-ce docker-ce-cli containerd.io docker-buildx-plugin docker-compose-plugin
diff --git a/.gitignore b/.gitignore
index e985f85..933a88b 100644
--- a/.gitignore
+++ b/.gitignore
@@ -5,6 +5,7 @@
.cache/
/bin
coverage.txt
+/test/e2e/dist/
/test/plugins/workspace
/test/plugins/dist/
*.pb.go
diff --git a/CHANGES.md b/CHANGES.md
index c628fe4..cd86291 100644
--- a/CHANGES.md
+++ b/CHANGES.md
@@ -7,6 +7,7 @@ Release Notes.
#### Features
* Support Windows plugin test.
+* Support Kafka reporter.
#### Plugins
diff --git a/Makefile b/Makefile
index b6c2385..b08b3dd 100644
--- a/Makefile
+++ b/Makefile
@@ -87,7 +87,7 @@ test: ## Run E2E scenario tests
lint: linter ## Run golangci-lint linter
@$(LOG_TARGET)
@for dir in $$(find . -name go.mod -exec dirname {} \; ); do \
- if [[ $$dir == "./test/plugins/scenarios/"* ]]; then \
+ if [[ $$dir == "./test/plugins/scenarios/"* ]] || [[ $$dir == "./test/e2e/"* ]]; then \
continue; \
fi; \
echo "Linting $$dir"; \
diff --git a/agent/reporter/imports.go b/agent/reporter/imports.go
index 67a1364..2b78317 100644
--- a/agent/reporter/imports.go
+++ b/agent/reporter/imports.go
@@ -27,6 +27,7 @@ import (
_ "os"
_ "strconv"
_ "strings"
+ _ "sync"
_ "time"
// imports the logs for reporter
@@ -55,6 +56,11 @@ import (
_ "google.golang.org/grpc/stats"
_ "google.golang.org/grpc/status"
+ // imports required packages for Kafka reporter
+ _ "github.com/segmentio/kafka-go"
+ _ "github.com/segmentio/kafka-go/compress"
+ _ "google.golang.org/protobuf/proto"
+
// imports protocols between agent and backend
_ "skywalking.apache.org/repo/goapi/collect/agent/configuration/v3"
_ "skywalking.apache.org/repo/goapi/collect/common/v3"
diff --git a/docs/en/advanced-features/kafka-reporter.md b/docs/en/advanced-features/kafka-reporter.md
new file mode 100644
index 0000000..fab6752
--- /dev/null
+++ b/docs/en/advanced-features/kafka-reporter.md
@@ -0,0 +1,148 @@
+# Kafka Reporter
+
+This document describes how to configure and use the Kafka reporter in the Apache SkyWalking Go agent. The Kafka reporter provides an alternative to the default gRPC reporter, allowing you to send trace, metrics, and log data to Apache Kafka.
+
+## Overview
+
+The SkyWalking Go agent can be configured to report collected telemetry data (traces, metrics, logs) to a Kafka cluster. This is useful for scenarios where Kafka is already part of your infrastructure or when you prefer Kafka's buffering and scalability features for handling observability data.
+
+**Note:** Even when the primary data reporting is set to Kafka (`reporter.type: kafka`), the CDS functionality itself still relies on gRPC communication with the SkyWalking OAP (Observability Analysis Platform). Therefore, you **must** also configure the relevant gRPC settings under `reporter.grpc` (or their corresponding environment variables like `SW_AGENT_REPORTER_GRPC_BACKEND_SERVICE`) for CDS to work correctly.
+
+
+## Enabling Kafka Reporter
+
+You can enable the Kafka reporter either through environment variables or by configuring the `agent.default.yaml` file.
+
+**Using Environment Variables:**
+
+Set the `SW_AGENT_REPORTER_TYPE` environment variable to `kafka`:
+```bash
+export SW_AGENT_REPORTER_TYPE=kafka
+```
+
+**Using `agent.default.yaml`:**
+
+Modify the `reporter.type` setting in your `agent.default.yaml` configuration file:
+```yaml
+reporter:
+ type: kafka # or grpc
+ # ... other global reporter settings
+```
+
+## Configuration
+
+The Kafka reporter requires specific configurations for connecting to your Kafka cluster and specifying topics for different data types. These can be set via environment variables or in the `agent.default.yaml` file. Environment variable names generally follow the pattern `SW_AGENT_REPORTER_KAFKA_OPTION_NAME_IN_UPPERCASE`.
+
+### Core Kafka Configuration
+
+These settings are typically found under the `reporter.kafka` section in `agent.default.yaml` or can be set using the corresponding environment variables.
+
+* **Broker Addresses:**
+ A comma-separated list of Kafka broker addresses.
+ * Environment Variable: `SW_AGENT_REPORTER_KAFKA_BROKERS`
+ * YAML: `reporter.kafka.brokers`
+ * Example: `kafka1:9092,kafka2:9092`
+
+
+* **Topic for Segments:**
+ The Kafka topic where trace segments will be sent.
+ * Environment Variable: `SW_AGENT_REPORTER_KAFKA_TOPIC_SEGMENT`
+ * YAML: `reporter.kafka.topic_segment`
+ * Example: `skywalking-segments`
+
+
+* **Topic for Metrics:**
+ The Kafka topic where metrics data will be sent.
+ * Environment Variable: `SW_AGENT_REPORTER_KAFKA_TOPIC_METER`
+ * YAML: `reporter.kafka.topic_meter`
+ * Example: `skywalking-meters`
+
+
+* **Topic for Logs:**
+ The Kafka topic where log data will be sent.
+ * Environment Variable: `SW_AGENT_REPORTER_KAFKA_TOPIC_LOGGING`
+ * YAML: `reporter.kafka.topic_logging`
+ * Example: `skywalking-logs`
+
+
+* **Topic for Management:** (Optional)
+ The Kafka topic for management-related messages (e.g., potentially for configurations or commands in future use).
+ * Environment Variable: `SW_AGENT_REPORTER_KAFKA_TOPIC_MANAGEMENT`
+ * YAML: `reporter.kafka.topic_management`
+ * Example: `skywalking-management`
+
+
+### Advanced Kafka Configuration
+
+These options allow fine-tuning of the Kafka producer behavior. They can also be configured via environment variables or in `agent.default.yaml` under `reporter.kafka`.
+
+* **Send Queue Size (Tracing, Metrics, Logs):**
+ The maximum size of the internal queue for buffering data before sending for each data type.
+ * Environment Variable: `SW_AGENT_REPORTER_KAFKA_MAX_SEND_QUEUE`
+ * YAML: `reporter.kafka.max_send_queue`
+ * Default: `5000`
+
+
+* **Batch Size:**
+ The maximum number of messages batched before being sent to a partition.
+ * Environment Variable: `SW_AGENT_REPORTER_KAFKA_BATCH_SIZE`
+ * YAML: `reporter.kafka.batchSize`
+ * Default: `1000`
+
+
+* **Batch Bytes:**
+ The maximum total bytes batched before being sent to a partition.
+ * Environment Variable: `SW_AGENT_REPORTER_KAFKA_BATCH_BYTES`
+ * YAML: `reporter.kafka.batchBytes`
+ * Default: `1048576` (1MB)
+
+
+* **Batch Timeout:**
+ The maximum time the producer will wait before sending a batch, even if `batchSize` or `batchBytes` is not met. (e.g., "1s", "500ms").
+ * Environment Variable: `SW_AGENT_REPORTER_KAFKA_BATCH_TIMEOUT`
+ * YAML: `reporter.kafka.batchTimeout`
+ * Default: `1s` (1 second)
+
+
+* **Acknowledgement (Acks):**
+ The level of acknowledgement required from Kafka brokers.
+ * `0`: No acknowledgement (producer does not wait).
+ * `1`: Leader acknowledgement (producer waits for the leader to write the record).
+ * `-1`: All in-sync replicas acknowledgement.
+ * Environment Variable: `SW_AGENT_REPORTER_KAFKA_ACKS`
+ * YAML: `reporter.kafka.acks`
+ * Default: `1` (leader)
+
+
+### Example `agent.default.yaml` Snippet for Kafka:
+
+```yaml
+reporter:
+ # Reporter type: "grpc" or "kafka"
+ type: ${SW_AGENT_REPORTER_TYPE:kafka}
+ kafka:
+ # Kafka broker addresses, comma separated
+ brokers: ${SW_AGENT_REPORTER_KAFKA_BROKERS:127.0.0.1:9092}
+ # Topic for segments data
+ topic_segment: ${SW_AGENT_REPORTER_KAFKA_TOPIC_SEGMENT:skywalking-segments}
+ # Topic for meters data
+ topic_meter: ${SW_AGENT_REPORTER_KAFKA_TOPIC_METER:skywalking-meters}
+ # Topic for logging data
+ topic_logging: ${SW_AGENT_REPORTER_KAFKA_TOPIC_LOGGING:skywalking-logs}
+ # Topic for management data
+ topic_management: ${SW_AGENT_REPORTER_KAFKA_TOPIC_MANAGEMENT:skywalking-managements}
+ # Send queue size
+ max_send_queue: ${SW_AGENT_REPORTER_KAFKA_MAX_SEND_QUEUE:5000}
+ # Batch size
+ batch_size: ${SW_AGENT_REPORTER_KAFKA_BATCH_SIZE:1000}
+ # Batch bytes
+ batch_bytes: ${SW_AGENT_REPORTER_KAFKA_BATCH_BYTES:1048576}
+ # Batch timeout millis
+ batch_timeout_millis: ${SW_AGENT_REPORTER_KAFKA_BATCH_TIMEOUT_MILLIS:1000}
+ # Acknowledge, 0: none, 1: leader, -1: all
+ acks: ${SW_AGENT_REPORTER_KAFKA_ACKS:1}
+```
+
+## Data Format
+
+The agent transforms the collected spans, metrics, and logs into the standard Apache SkyWalking data format before sending them to Kafka. The SkyWalking OAP (Observability Analysis Platform) should be configured with Kafka fetchers to consume and process this data from the specified Kafka topics. Ensure your OAP version is compatible and configured correctly to ingest data reported via Kafka.
\ No newline at end of file
diff --git a/docs/menu.yml b/docs/menu.yml
index 86c1f51..520b844 100644
--- a/docs/menu.yml
+++ b/docs/menu.yml
@@ -42,6 +42,8 @@ catalog:
path: /en/agent/plugin-configurations
- name: Transport Layer Security (TLS)
path: /en/advanced-features/grpc-tls
+ - name: Kafka Reporter
+ path: /en/advanced-features/kafka-reporter
- name: Manual APIs
catalog:
- name: Tracing APIs
diff --git a/go.work b/go.work
index 320599e..4afcbca 100644
--- a/go.work
+++ b/go.work
@@ -36,6 +36,9 @@ use (
./test/plugins/runner-helper
// define the plugin test scenarios
+ ./test/e2e/base/consumer
+ ./test/e2e/base/provider
+
./test/plugins/scenarios/dubbo
./test/plugins/scenarios/gin
./test/plugins/scenarios/short_versions_gin
diff --git a/go.work.sum b/go.work.sum
index 8955c42..724f218 100644
--- a/go.work.sum
+++ b/go.work.sum
@@ -508,6 +508,7 @@ github.com/apache/arrow/go/v11 v11.0.0/go.mod h1:Eg5OsL5H+e299f7u5ssuXsuHQVEGC4x
github.com/apache/dubbo-getty v1.4.9-0.20221022181821-4dc6252ce98c h1:2LE4IlyVBBlMo0ZDI+vq9YIb35dyij1YR5EnNWVPnNQ=
github.com/apache/dubbo-go-hessian2 v1.11.5 h1:rcK22+yMw2Hejm6GRG7WrdZ0DinW2QMZc01c7YVZjcQ=
github.com/apache/skywalking-go v0.0.0-20230411034404-b9270e98036b/go.mod h1:K0KR1RwXnGRlhw/ANpJ4Ex2CzhT4ST78MxmgHNi3/Z4=
+github.com/apache/skywalking-go v0.6.0/go.mod h1:vrfjgJ+a1R4N8v3s/1w7IuZrQoXsREnuZUQef+bzJ9c=
github.com/apache/thrift v0.13.0 h1:5hryIiq9gtn+MiLVn0wP37kb/uTeRZgN08WoCsAhIhI=
github.com/apache/thrift v0.16.0 h1:qEy6UW60iVOlUy+b9ZR0d5WzUWYGOo4HfopoyBaNmoY=
github.com/apache/thrift v0.16.0/go.mod h1:PHK3hniurgQaNMZYaCLEqXKsYK8upmhPbmdP2FXSqgU=
@@ -904,8 +905,6 @@ github.com/go-kratos/swagger-api v1.0.0 h1:R1JcAqLSvfDzqHoDnpK3Jt1MEJAXGNifV7AnB
github.com/go-ldap/ldap v3.0.2+incompatible h1:kD5HQcAzlQ7yrhfn+h+MSABeAy/jAJhvIJ/QDllP44g=
github.com/go-ldap/ldap/v3 v3.1.10 h1:7WsKqasmPThNvdl0Q5GPpbTDD/ZD98CfuawrMIuh7qQ=
github.com/go-logr/logr v0.2.0/go.mod h1:z6/tIYblkpsD+a4lm/fGIIU9mZ+XfAiaFtq7xTgseGU=
-github.com/go-logr/logr v1.2.3 h1:2DntVwHkVopvECVRSlL5PSo9eG+cAkDCuckLubN+rq0=
-github.com/go-logr/stdr v1.2.2 h1:hSWxHoqTgW2S2qGc0LTAI563KZ5YKYRhT3MFKZMbjag=
github.com/go-micro/plugins/v4/auth/jwt v1.2.0 h1:+dJHFnobRhX3QvTlHKCj5Sn2eeFsfNFzmFCc4HX5ULQ=
github.com/go-micro/plugins/v4/auth/jwt v1.2.0/go.mod h1:6tHH9WrpbEwmNQKJj8DutZtZdvY9gl0XBRJPnzEwl6Q=
github.com/go-micro/plugins/v4/broker/nats v1.1.0 h1:JCo2JksH0M6HC8q8L+x9QCqGRr5M0BADQw2Q50TWCas=
@@ -1223,7 +1222,6 @@ github.com/lyft/protoc-gen-validate v0.0.13 h1:KNt/RhmQTOLr7Aj8PsJ7mTronaFyx80mR
github.com/magiconair/properties v1.8.0/go.mod h1:PppfXfuXeibc/6YijjN8zIbojt8czPbwD3XqdrwzmxQ=
github.com/magiconair/properties v1.8.4/go.mod h1:y3VJvCyxH9uVvJTWEGAELF3aiYNyPKd5NZ3oSwXrF60=
github.com/magiconair/properties v1.8.6/go.mod h1:y3VJvCyxH9uVvJTWEGAELF3aiYNyPKd5NZ3oSwXrF60=
-github.com/magiconair/properties v1.8.7 h1:IeQXZAiQcpL9mgcAe1Nu6cX9LLw6ExEHKjN0VQdvPDY=
github.com/mailru/easyjson v0.7.0/go.mod h1:KAzv3t3aY1NaHWoQz1+4F1ccyAH66Jk7yos7ldAVICs=
github.com/marstr/guid v1.1.0 h1:/M4H/1G4avsieL6BbUwCOBzulmoeKVP5ux/3mQNnbyI=
github.com/marstr/guid v1.1.0/go.mod h1:74gB1z2wpxxInTG6yaqA7KrtM0NZ+RbrcqDvYHefzho=
@@ -1355,7 +1353,6 @@ github.com/oklog/oklog v0.3.2 h1:wVfs8F+in6nTBMkA7CbRw+zZMIB7nNM825cM1wuzoTk=
github.com/oklog/run v1.0.0 h1:Ru7dDtJNOyC66gQ5dQmaCa0qIsAUFY3sFpK1Xk8igrw=
github.com/oklog/ulid v1.3.1 h1:EGfNDEx6MqHz8B3uNV6QAib1UR2Lm97sHi3ocA6ESJ4=
github.com/olekukonko/tablewriter v0.0.0-20170122224234-a0225b3f23b5 h1:58+kh9C6jJVXYjt8IE48G2eWl6BjwU5Gj0gqY84fy78=
-github.com/olekukonko/tablewriter v0.0.5 h1:P2Ga83D34wi1o9J6Wh1mRuqd4mF/x/lgBS7N7AbDhec=
github.com/oliveagle/jsonpath v0.0.0-20180606110733-2e52cf6e6852 h1:Yl0tPBa8QPjGmesFh1D0rDy+q1Twx6FyU7VWHi8wZbI=
github.com/onsi/ginkgo v0.0.0-20151202141238-7f8ab55aaf3b/go.mod h1:lLunBs/Ym6LB5Z9jYTR76FiuTmxDTDusOGeTQH+WWjE=
github.com/onsi/ginkgo v1.10.1/go.mod h1:lLunBs/Ym6LB5Z9jYTR76FiuTmxDTDusOGeTQH+WWjE=
@@ -1543,7 +1540,6 @@ github.com/stretchr/objx v0.0.0-20180129172003-8a3f7159479f/go.mod h1:HFkY916IF+
github.com/stretchr/objx v0.3.0/go.mod h1:qt09Ya8vawLte6SNmTgCsAVtYtaKzEcn8ATUoHMkEqE=
github.com/stretchr/objx v0.5.0 h1:1zr/of2m5FGMsad5YfcqgdqdWrIhu+EBEJRhR1U7z/c=
github.com/stretchr/objx v0.5.2 h1:xuMeJ0Sdp5ZMRXx/aWO6RZxdr3beISkG5/G/aIRr3pY=
-github.com/stretchr/objx v0.5.2/go.mod h1:FRsXN1f5AsAjCGJKqEizvkpNtU+EGNCLh3NxZ/8L+MA=
github.com/stretchr/testify v0.0.0-20180303142811-b89eecf5ca5d/go.mod h1:a8OnRcib4nhh0OaRAV+Yts87kKdq0PP7pXfy6kDkUVs=
github.com/stretchr/testify v1.8.3 h1:RP3t2pwF7cMEbC1dqtB6poj3niw/9gnV4Cjg5oW5gtY=
github.com/stretchr/testify v1.8.3/go.mod h1:sz/lmYIOXD/1dqDmKjjqLyZ2RngseejIcXlSw2iwfAo=
diff --git a/plugins/core/go.mod b/plugins/core/go.mod
index 221ddde..0b59f06 100644
--- a/plugins/core/go.mod
+++ b/plugins/core/go.mod
@@ -8,6 +8,7 @@ require (
github.com/pkg/errors v0.9.1
github.com/stretchr/testify v1.8.2
google.golang.org/grpc v1.55.0
+ github.com/segmentio/kafka-go v0.4.47
skywalking.apache.org/repo/goapi v0.0.0-20230314034821-0c5a44bb767a
)
diff --git a/plugins/core/metrics.go b/plugins/core/metrics.go
index db97404..5e0fd8a 100644
--- a/plugins/core/metrics.go
+++ b/plugins/core/metrics.go
@@ -32,6 +32,9 @@ func (t *Tracer) Metrics() interface{} {
}
func (t *Tracer) initMetricsCollect(meterCollectSecond int) {
+ if meterCollectSecond < 0 {
+ return
+ }
collectDuration := time.Duration(meterCollectSecond) * time.Second
go func() {
for {
diff --git a/plugins/core/reporter/cds_manager.go b/plugins/core/reporter/cds_manager.go
new file mode 100644
index 0000000..467fe66
--- /dev/null
+++ b/plugins/core/reporter/cds_manager.go
@@ -0,0 +1,96 @@
+// 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 reporter
+
+import (
+ "context"
+ "time"
+
+ "github.com/apache/skywalking-go/plugins/core/operator"
+
+ configuration "skywalking.apache.org/repo/goapi/collect/agent/configuration/v3"
+)
+
+type CDSManager struct {
+ logger operator.LogOperator
+ serverAddr string
+ cdsInterval time.Duration
+ cdsService *ConfigDiscoveryService
+ cdsClient configuration.ConfigurationDiscoveryServiceClient
+ connManager *ConnectionManager
+ entity *Entity
+}
+
+func NewCDSManager(logger operator.LogOperator, serverAddr string, cdsInterval time.Duration, connManager *ConnectionManager) (*CDSManager, error) {
+ cds := &CDSManager{
+ logger: logger,
+ serverAddr: serverAddr,
+ cdsInterval: cdsInterval,
+ connManager: connManager,
+ }
+ if cdsInterval > 0 {
+ conn, err := connManager.GetConnection(serverAddr)
+ if err != nil {
+ return nil, err
+ }
+ cds.cdsClient = configuration.NewConfigurationDiscoveryServiceClient(conn)
+ cds.cdsService = NewConfigDiscoveryService()
+ }
+ return cds, nil
+}
+
+func (r *CDSManager) InitCDS(entity *Entity, cdsWatchers []AgentConfigChangeWatcher) {
+ if r.cdsClient == nil {
+ return
+ }
+ r.entity = entity
+
+ // bind watchers
+ r.cdsService.BindWatchers(cdsWatchers)
+
+ // fetch config
+ go func() {
+ for {
+ switch r.connManager.GetConnectionStatus(r.serverAddr) {
+ case ConnectionStatusShutdown:
+ break
+ case ConnectionStatusDisconnect:
+ time.Sleep(r.cdsInterval)
+ continue
+ }
+
+ configurations, err := r.cdsClient.FetchConfigurations(context.Background(), &configuration.ConfigurationSyncRequest{
+ Service: r.entity.ServiceName,
+ Uuid: r.cdsService.UUID,
+ })
+
+ if err != nil {
+ r.logger.Errorf("fetch dynamic configuration error %v", err)
+ time.Sleep(r.cdsInterval)
+ continue
+ }
+
+ if len(configurations.GetCommands()) > 0 && configurations.GetCommands()[0].Command == "ConfigurationDiscoveryCommand" {
+ command := configurations.GetCommands()[0]
+ r.cdsService.HandleCommand(command)
+ }
+
+ time.Sleep(r.cdsInterval)
+ }
+ }()
+}
diff --git a/plugins/core/reporter/conn_manager.go b/plugins/core/reporter/conn_manager.go
new file mode 100644
index 0000000..2968511
--- /dev/null
+++ b/plugins/core/reporter/conn_manager.go
@@ -0,0 +1,218 @@
+// 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 reporter
+
+import (
+ "crypto/tls"
+ "crypto/x509"
+ "fmt"
+ "os"
+ "sync"
+ "time"
+
+ "google.golang.org/grpc/connectivity"
+
+ "google.golang.org/grpc"
+ "google.golang.org/grpc/backoff"
+ "google.golang.org/grpc/credentials"
+ "google.golang.org/grpc/credentials/insecure"
+ "google.golang.org/grpc/metadata"
+
+ "github.com/apache/skywalking-go/plugins/core/operator"
+)
+
+var authKey = "Authentication"
+
+func NewConnectionManager(logger operator.LogOperator, checkInterval time.Duration,
+ serverAddr string, auth string, creds credentials.TransportCredentials) (*ConnectionManager, error) {
+ c := &ConnectionManager{
+ logger: logger,
+ checkInterval: checkInterval,
+ serverAddr: serverAddr,
+ md: metadata.New(map[string]string{authKey: auth}),
+ creds: creds,
+ connManager: make(map[string]*ManagedConnection),
+ mu: sync.RWMutex{},
+ }
+ return c, nil
+}
+
+type ConnectionManager struct {
+ logger operator.LogOperator
+ checkInterval time.Duration
+ serverAddr string
+ md metadata.MD
+ creds credentials.TransportCredentials
+ connManager map[string]*ManagedConnection
+ mu sync.RWMutex
+}
+
+type ManagedConnection struct {
+ connection *grpc.ClientConn
+ status ConnectionStatus
+ refCount int
+}
+
+func (cm *ConnectionManager) GetMD() metadata.MD {
+ return cm.md
+}
+
+func (cm *ConnectionManager) GetConnection(serverAddr string) (*grpc.ClientConn, error) {
+ managed, exists := cm.connManager[serverAddr]
+ if exists {
+ managed.refCount++
+ return managed.connection, nil
+ }
+ conn, err := cm.createConnection()
+ if err != nil {
+ return nil, err
+ }
+ managed = &ManagedConnection{
+ connection: conn,
+ status: ConnectionStatusConnected,
+ refCount: 1,
+ }
+ cm.connManager[serverAddr] = managed
+ go cm.checkConnectionStatus(serverAddr)
+ return conn, nil
+}
+
+func (cm *ConnectionManager) createConnection() (*grpc.ClientConn, error) {
+ var credsDialOption grpc.DialOption
+ if cm.creds != nil {
+ // use tls
+ credsDialOption = grpc.WithTransportCredentials(cm.creds)
+ } else {
+ credsDialOption = grpc.WithTransportCredentials(insecure.NewCredentials())
+ }
+
+ conn, err := grpc.Dial(cm.serverAddr, credsDialOption, grpc.WithConnectParams(grpc.ConnectParams{
+ // update the max backoff delay interval
+ Backoff: backoff.Config{
+ BaseDelay: 1.0 * time.Second,
+ Multiplier: 1.6,
+ Jitter: 0.2,
+ MaxDelay: cm.checkInterval,
+ },
+ }))
+ return conn, err
+}
+
+func (cm *ConnectionManager) checkConnectionStatus(serverAddr string) {
+ for {
+ cm.mu.Lock()
+ managed, exists := cm.connManager[serverAddr]
+ cm.mu.Unlock()
+ if !exists {
+ return
+ }
+ state := managed.connection.GetState()
+ var newStatus ConnectionStatus
+ switch state {
+ case connectivity.TransientFailure:
+ newStatus = ConnectionStatusDisconnect
+ case connectivity.Shutdown:
+ newStatus = ConnectionStatusShutdown
+ default:
+ newStatus = ConnectionStatusConnected
+ }
+ if newStatus != managed.status {
+ cm.mu.Lock()
+ managed.status = newStatus
+ cm.mu.Unlock()
+ }
+ time.Sleep(5 * time.Second)
+ }
+}
+
+func (cm *ConnectionManager) ReleaseConnection(serverAddr string) error {
+ cm.mu.Lock()
+ defer cm.mu.Unlock()
+ managed, exists := cm.connManager[serverAddr]
+ if !exists {
+ return nil
+ }
+ managed.refCount--
+ if managed.refCount <= 0 {
+ if err := managed.connection.Close(); err != nil {
+ cm.logger.Error(err)
+ }
+ delete(cm.connManager, serverAddr)
+ }
+ return nil
+}
+
+func (cm *ConnectionManager) GetConnectionStatus(serverAddr string) ConnectionStatus {
+ cm.mu.Lock()
+ defer cm.mu.Unlock()
+ managed, exists := cm.connManager[serverAddr]
+ if !exists {
+ return ConnectionStatusShutdown
+ }
+ return managed.status
+}
+
+// nolint
+func generateTLSCredential(caPath, clientKeyPath, clientCertChainPath string, skipVerify bool) (tc credentials.TransportCredentials, tlsErr error) {
+ if err := checkTLSFile(caPath); err != nil {
+ return nil, err
+ }
+ tlsConfig := new(tls.Config)
+ tlsConfig.Renegotiation = tls.RenegotiateNever
+ tlsConfig.InsecureSkipVerify = skipVerify
+ caPem, err := os.ReadFile(caPath)
+ if err != nil {
+ return nil, err
+ }
+ certPool := x509.NewCertPool()
+ if !certPool.AppendCertsFromPEM(caPem) {
+ return nil, fmt.Errorf("failed to append certificates")
+ }
+ tlsConfig.RootCAs = certPool
+
+ if clientKeyPath != "" && clientCertChainPath != "" {
+ if err := checkTLSFile(clientKeyPath); err != nil {
+ return nil, err
+ }
+ if err := checkTLSFile(clientCertChainPath); err != nil {
+ return nil, err
+ }
+ clientPem, err := tls.LoadX509KeyPair(clientCertChainPath, clientKeyPath)
+ if err != nil {
+ return nil, err
+ }
+ tlsConfig.Certificates = []tls.Certificate{clientPem}
+ }
+ return credentials.NewTLS(tlsConfig), nil
+}
+
+// checkTLSFile checks the TLS files.
+func checkTLSFile(path string) error {
+ file, err := os.Open(path)
+ if err != nil {
+ return err
+ }
+ stat, err := file.Stat()
+ if err != nil {
+ return err
+ }
+ if stat.Size() == 0 {
+ return fmt.Errorf("the TLS file is illegal: %s", path)
+ }
+ return nil
+}
diff --git a/plugins/core/reporter/grpc/grpc.go b/plugins/core/reporter/grpc/grpc.go
index adbf0a4..b2132ef 100644
--- a/plugins/core/reporter/grpc/grpc.go
+++ b/plugins/core/reporter/grpc/grpc.go
@@ -22,14 +22,8 @@ import (
"io"
"time"
- "google.golang.org/grpc"
- "google.golang.org/grpc/backoff"
- "google.golang.org/grpc/connectivity"
- "google.golang.org/grpc/credentials"
- "google.golang.org/grpc/credentials/insecure"
"google.golang.org/grpc/metadata"
- configuration "skywalking.apache.org/repo/goapi/collect/agent/configuration/v3"
agentv3 "skywalking.apache.org/repo/goapi/collect/language/agent/v3"
logv3 "skywalking.apache.org/repo/goapi/collect/logging/v3"
managementv3 "skywalking.apache.org/repo/goapi/collect/management/v3"
@@ -39,151 +33,80 @@ import (
)
const (
- maxSendQueueSize int32 = 30000
- defaultCheckInterval = 20 * time.Second
- defaultCDSInterval = 20 * time.Second
+ maxSendQueueSize int32 = 30000
)
// NewGRPCReporter create a new reporter to send data to gRPC oap server. Only one backend address is allowed.
-func NewGRPCReporter(logger operator.LogOperator, serverAddr string, opts ...ReporterOption) (reporter.Reporter, error) {
+func NewGRPCReporter(logger operator.LogOperator,
+ serverAddr string,
+ checkInterval time.Duration,
+ connManager *reporter.ConnectionManager,
+ cdsManager *reporter.CDSManager,
+ opts ...ReporterOption,
+) (reporter.Reporter, error) {
r := &gRPCReporter{
- logger: logger,
- tracingSendCh: make(chan *agentv3.SegmentObject, maxSendQueueSize),
- metricsSendCh: make(chan []*agentv3.MeterData, maxSendQueueSize),
- logSendCh: make(chan *logv3.LogData, maxSendQueueSize),
- checkInterval: defaultCheckInterval,
- cdsInterval: defaultCDSInterval, // cds default on
- connectionStatus: reporter.ConnectionStatusConnected,
+ logger: logger,
+ serverAddr: serverAddr,
+ tracingSendCh: make(chan *agentv3.SegmentObject, maxSendQueueSize),
+ metricsSendCh: make(chan []*agentv3.MeterData, maxSendQueueSize),
+ logSendCh: make(chan *logv3.LogData, maxSendQueueSize),
+ checkInterval: checkInterval,
+ connManager: connManager,
+ cdsManager: cdsManager,
}
for _, o := range opts {
o(r)
}
- var credsDialOption grpc.DialOption
- if r.creds != nil {
- // use tls
- credsDialOption = grpc.WithTransportCredentials(r.creds)
- } else {
- credsDialOption = grpc.WithTransportCredentials(insecure.NewCredentials())
- }
-
- conn, err := grpc.Dial(serverAddr, credsDialOption, grpc.WithConnectParams(grpc.ConnectParams{
- // update the max backoff delay interval
- Backoff: backoff.Config{
- BaseDelay: 1.0 * time.Second,
- Multiplier: 1.6,
- Jitter: 0.2,
- MaxDelay: r.checkInterval,
- },
- }))
+ conn, err := connManager.GetConnection(serverAddr)
if err != nil {
return nil, err
}
- r.conn = conn
- r.traceClient = agentv3.NewTraceSegmentReportServiceClient(r.conn)
- r.metricsClient = agentv3.NewMeterReportServiceClient(r.conn)
- r.logClient = logv3.NewLogReportServiceClient(r.conn)
- r.managementClient = managementv3.NewManagementServiceClient(r.conn)
- if r.cdsInterval > 0 {
- r.cdsClient = configuration.NewConfigurationDiscoveryServiceClient(r.conn)
- r.cdsService = reporter.NewConfigDiscoveryService()
- }
+ r.traceClient = agentv3.NewTraceSegmentReportServiceClient(conn)
+ r.metricsClient = agentv3.NewMeterReportServiceClient(conn)
+ r.logClient = logv3.NewLogReportServiceClient(conn)
+ r.managementClient = managementv3.NewManagementServiceClient(conn)
return r, nil
}
type gRPCReporter struct {
entity *reporter.Entity
+ serverAddr string
logger operator.LogOperator
tracingSendCh chan *agentv3.SegmentObject
metricsSendCh chan []*agentv3.MeterData
logSendCh chan *logv3.LogData
- conn *grpc.ClientConn
traceClient agentv3.TraceSegmentReportServiceClient
metricsClient agentv3.MeterReportServiceClient
logClient logv3.LogReportServiceClient
managementClient managementv3.ManagementServiceClient
checkInterval time.Duration
- cdsInterval time.Duration
- cdsService *reporter.ConfigDiscoveryService
- cdsClient configuration.ConfigurationDiscoveryServiceClient
-
- md metadata.MD
- creds credentials.TransportCredentials
// bootFlag is set if Boot be executed
- bootFlag bool
- connectionStatus reporter.ConnectionStatus
+ bootFlag bool
+ transform *reporter.Transform
+ connManager *reporter.ConnectionManager
+ cdsManager *reporter.CDSManager
}
func (r *gRPCReporter) Boot(entity *reporter.Entity, cdsWatchers []reporter.AgentConfigChangeWatcher) {
r.entity = entity
+ r.transform = reporter.NewTransform(entity)
r.initSendPipeline()
r.check()
- r.initCDS(cdsWatchers)
+ r.cdsManager.InitCDS(entity, cdsWatchers)
r.bootFlag = true
}
func (r *gRPCReporter) ConnectionStatus() reporter.ConnectionStatus {
- return r.connectionStatus
+ return r.connManager.GetConnectionStatus(r.serverAddr)
}
func (r *gRPCReporter) SendTracing(spans []reporter.ReportedSpan) {
- spanSize := len(spans)
- if spanSize < 1 {
+ segmentObject := r.transform.TransformSegmentObject(spans)
+ if segmentObject == nil {
return
}
- rootSpan := spans[spanSize-1]
- rootCtx := rootSpan.Context()
- segmentObject := &agentv3.SegmentObject{
- TraceId: rootCtx.GetTraceID(),
- TraceSegmentId: rootCtx.GetSegmentID(),
- Spans: make([]*agentv3.SpanObject, spanSize),
- Service: r.entity.ServiceName,
- ServiceInstance: r.entity.ServiceInstanceName,
- }
- for i, s := range spans {
- spanCtx := s.Context()
- segmentObject.Spans[i] = &agentv3.SpanObject{
- SpanId: spanCtx.GetSpanID(),
- ParentSpanId: spanCtx.GetParentSpanID(),
- StartTime: s.StartTime(),
- EndTime: s.EndTime(),
- OperationName: s.OperationName(),
- Peer: s.Peer(),
- SpanType: s.SpanType(),
- SpanLayer: s.SpanLayer(),
- ComponentId: s.ComponentID(),
- IsError: s.IsError(),
- Tags: s.Tags(),
- Logs: s.Logs(),
- }
- srr := make([]*agentv3.SegmentReference, 0)
- if i == (spanSize-1) && spanCtx.GetParentSpanID() > -1 {
- srr = append(srr, &agentv3.SegmentReference{
- RefType: agentv3.RefType_CrossThread,
- TraceId: spanCtx.GetTraceID(),
- ParentTraceSegmentId: spanCtx.GetParentSegmentID(),
- ParentSpanId: spanCtx.GetParentSpanID(),
- ParentService: r.entity.ServiceName,
- ParentServiceInstance: r.entity.ServiceInstanceName,
- })
- }
- if len(s.Refs()) > 0 {
- for _, tc := range s.Refs() {
- srr = append(srr, &agentv3.SegmentReference{
- RefType: agentv3.RefType_CrossProcess,
- TraceId: spanCtx.GetTraceID(),
- ParentTraceSegmentId: tc.GetParentSegmentID(),
- ParentSpanId: tc.GetParentSpanID(),
- ParentService: tc.GetParentService(),
- ParentServiceInstance: tc.GetParentServiceInstance(),
- ParentEndpoint: tc.GetParentEndpoint(),
- NetworkAddressUsedAtPeer: tc.GetAddressUsedAtClient(),
- })
- }
- }
- segmentObject.Spans[i].Refs = srr
- }
defer func() {
// recover the panic caused by close tracingSendCh
if err := recover(); err != nil {
@@ -198,47 +121,12 @@ func (r *gRPCReporter) SendTracing(spans []reporter.ReportedSpan) {
}
func (r *gRPCReporter) SendMetrics(metrics []reporter.ReportedMeter) {
- if len(metrics) == 0 {
+ meters := r.transform.TransformMeterData(metrics)
+ if meters == nil {
return
}
- meters := make([]*agentv3.MeterData, len(metrics))
- for i, m := range metrics {
- meter := &agentv3.MeterData{}
- switch data := m.(type) {
- case reporter.ReportedMeterSingleValue:
- meter.Metric = &agentv3.MeterData_SingleValue{
- SingleValue: &agentv3.MeterSingleValue{
- Name: data.Name(),
- Labels: r.convertLabels(data.Labels()),
- Value: data.Value(),
- },
- }
- case reporter.ReportedMeterHistogram:
- buckets := make([]*agentv3.MeterBucketValue, len(data.BucketValues()))
- for i, b := range data.BucketValues() {
- buckets[i] = &agentv3.MeterBucketValue{
- Bucket: b.Bucket(),
- Count: b.Count(),
- IsNegativeInfinity: b.IsNegativeInfinity(),
- }
- }
- meter.Metric = &agentv3.MeterData_Histogram{
- Histogram: &agentv3.MeterHistogram{
- Name: data.Name(),
- Labels: r.convertLabels(data.Labels()),
- Values: buckets,
- },
- }
- }
-
- meters[i] = meter
- }
-
- meters[0].Service = r.entity.ServiceName
- meters[0].ServiceInstance = r.entity.ServiceInstanceName
- meters[0].Timestamp = time.Now().UnixNano() / int64(time.Millisecond)
defer func() {
- // recover the panic caused by close tracingSendCh
+ // recover the panic caused by close metricsSendCh
if err := recover(); err != nil {
r.logger.Errorf("reporter metrics err %v", err)
}
@@ -262,20 +150,6 @@ func (r *gRPCReporter) SendLog(log *logv3.LogData) {
}
}
-func (r *gRPCReporter) convertLabels(labels map[string]string) []*agentv3.Label {
- if len(labels) == 0 {
- return nil
- }
- ls := make([]*agentv3.Label, 0)
- for k, v := range labels {
- ls = append(ls, &agentv3.Label{
- Name: k,
- Value: v,
- })
- }
- return ls
-}
-
func (r *gRPCReporter) Close() {
if r.bootFlag {
if r.tracingSendCh != nil {
@@ -290,10 +164,8 @@ func (r *gRPCReporter) Close() {
}
func (r *gRPCReporter) closeGRPCConn() {
- if r.conn != nil {
- if err := r.conn.Close(); err != nil {
- r.logger.Error(err)
- }
+ if err := r.connManager.ReleaseConnection(r.serverAddr); err != nil {
+ r.logger.Error(err)
}
}
@@ -305,7 +177,7 @@ func (r *gRPCReporter) initSendPipeline() {
go func() {
StreamLoop:
for {
- switch r.updateConnectionStatus() {
+ switch r.connManager.GetConnectionStatus(r.serverAddr) {
case reporter.ConnectionStatusShutdown:
break
case reporter.ConnectionStatusDisconnect:
@@ -313,7 +185,7 @@ func (r *gRPCReporter) initSendPipeline() {
continue StreamLoop
}
- stream, err := r.traceClient.Collect(metadata.NewOutgoingContext(context.Background(), r.md))
+ stream, err := r.traceClient.Collect(metadata.NewOutgoingContext(context.Background(), r.connManager.GetMD()))
if err != nil {
r.logger.Errorf("open stream error %v", err)
time.Sleep(5 * time.Second)
@@ -335,7 +207,7 @@ func (r *gRPCReporter) initSendPipeline() {
go func() {
StreamLoop:
for {
- switch r.updateConnectionStatus() {
+ switch r.connManager.GetConnectionStatus(r.serverAddr) {
case reporter.ConnectionStatusShutdown:
break
case reporter.ConnectionStatusDisconnect:
@@ -343,7 +215,7 @@ func (r *gRPCReporter) initSendPipeline() {
continue StreamLoop
}
- stream, err := r.metricsClient.CollectBatch(metadata.NewOutgoingContext(context.Background(), r.md))
+ stream, err := r.metricsClient.CollectBatch(metadata.NewOutgoingContext(context.Background(), r.connManager.GetMD()))
if err != nil {
r.logger.Errorf("open stream error %v", err)
time.Sleep(5 * time.Second)
@@ -366,7 +238,7 @@ func (r *gRPCReporter) initSendPipeline() {
go func() {
StreamLoop:
for {
- switch r.updateConnectionStatus() {
+ switch r.connManager.GetConnectionStatus(r.serverAddr) {
case reporter.ConnectionStatusShutdown:
break
case reporter.ConnectionStatusDisconnect:
@@ -374,7 +246,7 @@ func (r *gRPCReporter) initSendPipeline() {
continue StreamLoop
}
- stream, err := r.logClient.Collect(metadata.NewOutgoingContext(context.Background(), r.md))
+ stream, err := r.logClient.Collect(metadata.NewOutgoingContext(context.Background(), r.connManager.GetMD()))
if err != nil {
r.logger.Errorf("open stream error %v", err)
time.Sleep(5 * time.Second)
@@ -394,59 +266,6 @@ func (r *gRPCReporter) initSendPipeline() {
}()
}
-func (r *gRPCReporter) updateConnectionStatus() reporter.ConnectionStatus {
- state := r.conn.GetState()
- switch state {
- case connectivity.TransientFailure:
- r.connectionStatus = reporter.ConnectionStatusDisconnect
- case connectivity.Shutdown:
- r.connectionStatus = reporter.ConnectionStatusShutdown
- default:
- r.connectionStatus = reporter.ConnectionStatusConnected
- }
- return r.connectionStatus
-}
-
-func (r *gRPCReporter) initCDS(cdsWatchers []reporter.AgentConfigChangeWatcher) {
- if r.cdsClient == nil {
- return
- }
-
- // bind watchers
- r.cdsService.BindWatchers(cdsWatchers)
-
- // fetch config
- go func() {
- for {
- switch r.updateConnectionStatus() {
- case reporter.ConnectionStatusShutdown:
- break
- case reporter.ConnectionStatusDisconnect:
- time.Sleep(r.cdsInterval)
- continue
- }
-
- configurations, err := r.cdsClient.FetchConfigurations(context.Background(), &configuration.ConfigurationSyncRequest{
- Service: r.entity.ServiceName,
- Uuid: r.cdsService.UUID,
- })
-
- if err != nil {
- r.logger.Errorf("fetch dynamic configuration error %v", err)
- time.Sleep(r.cdsInterval)
- continue
- }
-
- if len(configurations.GetCommands()) > 0 && configurations.GetCommands()[0].Command == "ConfigurationDiscoveryCommand" {
- command := configurations.GetCommands()[0]
- r.cdsService.HandleCommand(command)
- }
-
- time.Sleep(r.cdsInterval)
- }
- }()
-}
-
func (r *gRPCReporter) closeTracingStream(stream agentv3.TraceSegmentReportService_CollectClient) {
_, err := stream.CloseAndRecv()
if err != nil && err != io.EOF {
@@ -469,22 +288,24 @@ func (r *gRPCReporter) closeLogStream(stream logv3.LogReportService_CollectClien
}
func (r *gRPCReporter) reportInstanceProperties() (err error) {
- _, err = r.managementClient.ReportInstanceProperties(metadata.NewOutgoingContext(context.Background(), r.md), &managementv3.InstanceProperties{
- Service: r.entity.ServiceName,
- ServiceInstance: r.entity.ServiceInstanceName,
- Properties: r.entity.Props,
- })
+ _, err = r.managementClient.ReportInstanceProperties(
+ metadata.NewOutgoingContext(context.Background(), r.connManager.GetMD()),
+ &managementv3.InstanceProperties{
+ Service: r.entity.ServiceName,
+ ServiceInstance: r.entity.ServiceInstanceName,
+ Properties: r.entity.Props,
+ })
return err
}
func (r *gRPCReporter) check() {
- if r.checkInterval < 0 || r.conn == nil || r.managementClient == nil {
+ if r.checkInterval < 0 || r.managementClient == nil {
return
}
go func() {
instancePropertiesSubmitted := false
for {
- switch r.updateConnectionStatus() {
+ switch r.connManager.GetConnectionStatus(r.serverAddr) {
case reporter.ConnectionStatusShutdown:
break
case reporter.ConnectionStatusDisconnect:
@@ -502,10 +323,12 @@ func (r *gRPCReporter) check() {
instancePropertiesSubmitted = true
}
- _, err := r.managementClient.KeepAlive(metadata.NewOutgoingContext(context.Background(), r.md), &managementv3.InstancePingPkg{
- Service: r.entity.ServiceName,
- ServiceInstance: r.entity.ServiceInstanceName,
- })
+ _, err := r.managementClient.KeepAlive(
+ metadata.NewOutgoingContext(context.Background(), r.connManager.GetMD()),
+ &managementv3.InstancePingPkg{
+ Service: r.entity.ServiceName,
+ ServiceInstance: r.entity.ServiceInstanceName,
+ })
if err != nil {
r.logger.Errorf("send keep alive signal error %v", err)
diff --git a/plugins/core/reporter/grpc/grpc_opts_enhance.go b/plugins/core/reporter/grpc/grpc_opts_enhance.go
index db26648..e05a841 100644
--- a/plugins/core/reporter/grpc/grpc_opts_enhance.go
+++ b/plugins/core/reporter/grpc/grpc_opts_enhance.go
@@ -18,32 +18,14 @@
package grpc
import (
- "crypto/tls"
- "crypto/x509"
- "fmt"
- "os"
- "time"
-
- "google.golang.org/grpc/credentials"
- "google.golang.org/grpc/metadata"
-
agentv3 "skywalking.apache.org/repo/goapi/collect/language/agent/v3"
logv3 "skywalking.apache.org/repo/goapi/collect/logging/v3"
)
-var authKey = "Authentication"
-
// ReporterOption allows for functional options to adjust behavior
// of a gRPC reporter to be created by NewGRPCReporter
type ReporterOption func(r *gRPCReporter)
-// WithCheckInterval setup service and endpoint registry check interval
-func WithCheckInterval(interval time.Duration) ReporterOption {
- return func(r *gRPCReporter) {
- r.checkInterval = interval
- }
-}
-
// WithMaxSendQueueSize setup send span queue buffer length
func WithMaxSendQueueSize(maxSendQueueSize int) ReporterOption {
return func(r *gRPCReporter) {
@@ -54,72 +36,7 @@ func WithMaxSendQueueSize(maxSendQueueSize int) ReporterOption {
}
// WithTransportCredentials setup transport layer security
-func WithTransportCredentials(creds credentials.TransportCredentials) ReporterOption {
- return func(r *gRPCReporter) {
- r.creds = creds
- }
-}
// WithAuthentication used Authentication for gRPC
-func WithAuthentication(auth string) ReporterOption {
- return func(r *gRPCReporter) {
- r.md = metadata.New(map[string]string{authKey: auth})
- }
-}
// WithCDS setup Configuration Discovery Service to dynamic config
-func WithCDS(interval time.Duration) ReporterOption {
- return func(r *gRPCReporter) {
- r.cdsInterval = interval
- }
-}
-
-//nolint
-func generateTLSCredential(caPath, clientKeyPath, clientCertChainPath string, skipVerify bool) (tc credentials.TransportCredentials, tlsErr error) {
- if err := checkTLSFile(caPath); err != nil {
- return nil, err
- }
- tlsConfig := new(tls.Config)
- tlsConfig.Renegotiation = tls.RenegotiateNever
- tlsConfig.InsecureSkipVerify = skipVerify
- caPem, err := os.ReadFile(caPath)
- if err != nil {
- return nil, err
- }
- certPool := x509.NewCertPool()
- if !certPool.AppendCertsFromPEM(caPem) {
- return nil, fmt.Errorf("failed to append certificates")
- }
- tlsConfig.RootCAs = certPool
-
- if clientKeyPath != "" && clientCertChainPath != "" {
- if err := checkTLSFile(clientKeyPath); err != nil {
- return nil, err
- }
- if err := checkTLSFile(clientCertChainPath); err != nil {
- return nil, err
- }
- clientPem, err := tls.LoadX509KeyPair(clientCertChainPath, clientKeyPath)
- if err != nil {
- return nil, err
- }
- tlsConfig.Certificates = []tls.Certificate{clientPem}
- }
- return credentials.NewTLS(tlsConfig), nil
-}
-
-// checkTLSFile checks the TLS files.
-func checkTLSFile(path string) error {
- file, err := os.Open(path)
- if err != nil {
- return err
- }
- stat, err := file.Stat()
- if err != nil {
- return err
- }
- if stat.Size() == 0 {
- return fmt.Errorf("the TLS file is illegal: %s", path)
- }
- return nil
-}
diff --git a/plugins/core/reporter/kafka/kafka.go b/plugins/core/reporter/kafka/kafka.go
new file mode 100644
index 0000000..8e25d73
--- /dev/null
+++ b/plugins/core/reporter/kafka/kafka.go
@@ -0,0 +1,352 @@
+// 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 kafka
+
+import (
+ "context"
+ "strings"
+ "time"
+
+ "github.com/segmentio/kafka-go"
+ "github.com/segmentio/kafka-go/compress"
+ "google.golang.org/protobuf/proto"
+ agentv3 "skywalking.apache.org/repo/goapi/collect/language/agent/v3"
+ logv3 "skywalking.apache.org/repo/goapi/collect/logging/v3"
+ managementv3 "skywalking.apache.org/repo/goapi/collect/management/v3"
+
+ "github.com/apache/skywalking-go/plugins/core/operator"
+ "github.com/apache/skywalking-go/plugins/core/reporter"
+)
+
+const (
+ kafkaMaxSendQueueSize int32 = 30000
+ topicKeyRegister = "register-"
+)
+
+var internalReporterContextKey = context.Background()
+
+type kafkaReporter struct {
+ entity *reporter.Entity
+ logger operator.LogOperator
+ writer *kafka.Writer
+ brokerAddr []string
+ tracingSendCh chan *agentv3.SegmentObject
+ metricsSendCh chan []*agentv3.MeterData
+ logSendCh chan *logv3.LogData
+ bootFlag bool
+ checkInterval time.Duration
+ connectionStatus reporter.ConnectionStatus
+ topicSegment string
+ topicMeter string
+ topicLogging string
+ topicManagement string
+ transform *reporter.Transform
+ cdsManager *reporter.CDSManager
+}
+
+func NewKafkaReporter(logger operator.LogOperator,
+ brokers string,
+ checkInterval time.Duration,
+ cdsManager *reporter.CDSManager,
+ opts ...ReporterOptionKafka,
+) (reporter.Reporter, error) {
+ r := &kafkaReporter{
+ logger: logger,
+ tracingSendCh: make(chan *agentv3.SegmentObject, kafkaMaxSendQueueSize),
+ metricsSendCh: make(chan []*agentv3.MeterData, kafkaMaxSendQueueSize),
+ logSendCh: make(chan *logv3.LogData, kafkaMaxSendQueueSize),
+ checkInterval: checkInterval,
+ connectionStatus: reporter.ConnectionStatusDisconnect,
+ cdsManager: cdsManager,
+ }
+
+ r.brokerAddr = strings.Split(brokers, ",")
+ writer := &kafka.Writer{
+ Addr: kafka.TCP(r.brokerAddr...),
+ Balancer: &kafka.RoundRobin{},
+ MaxAttempts: 10,
+ BatchSize: 1000,
+ BatchBytes: 1048576,
+ BatchTimeout: 1000 * time.Millisecond,
+ RequiredAcks: kafka.RequireOne,
+ Async: false,
+ Compression: compress.None,
+ ErrorLogger: kafka.LoggerFunc(logger.Errorf),
+ AllowAutoTopicCreation: true,
+ }
+ r.writer = writer
+ for _, opt := range opts {
+ opt(r)
+ }
+ return r, nil
+}
+
+func (r *kafkaReporter) Boot(entity *reporter.Entity, cdsWatchers []reporter.AgentConfigChangeWatcher) {
+ r.entity = entity
+ r.transform = reporter.NewTransform(entity)
+ r.updateConnectionStatus()
+ r.initSendPipeline()
+ r.check()
+ r.cdsManager.InitCDS(entity, cdsWatchers)
+ r.bootFlag = true
+}
+
+func (r *kafkaReporter) updateConnectionStatus() {
+ if r.checkKafkaConnection() {
+ r.connectionStatus = reporter.ConnectionStatusConnected
+ } else {
+ r.connectionStatus = reporter.ConnectionStatusDisconnect
+ }
+}
+
+func (r *kafkaReporter) checkKafkaConnection() bool {
+ firstAddr := r.brokerAddr[0]
+ conn, err := kafka.Dial("tcp", firstAddr)
+ if err != nil {
+ r.logger.Errorf("kafka connection error %v", err)
+ return false
+ }
+ defer func() {
+ if err = conn.Close(); err != nil {
+ r.logger.Errorf("kafka connection close error %v", err)
+ }
+ }()
+ _, err = conn.Brokers()
+ if err != nil {
+ r.logger.Errorf("kafka connection error %v", err)
+ return false
+ }
+ return true
+}
+
+func (r *kafkaReporter) initSendPipeline() {
+ go r.tracingSendLoop()
+ go r.metricsSendLoop()
+ go r.logSendLoop()
+}
+
+func (r *kafkaReporter) tracingSendLoop() {
+ consecutiveErrors := 0
+ logFrequency := 30
+ for s := range r.tracingSendCh {
+ payload, err := proto.Marshal(s)
+ if err != nil {
+ r.logger.Errorf("marshal segment error %v", err)
+ continue
+ }
+ ctx := context.WithValue(context.Background(), internalReporterContextKey, true)
+ err = r.writer.WriteMessages(ctx, kafka.Message{
+ Topic: r.topicSegment,
+ Key: []byte(s.GetTraceSegmentId()),
+ Value: payload,
+ })
+ if err != nil {
+ consecutiveErrors++
+ if consecutiveErrors == 1 || consecutiveErrors%logFrequency == 0 {
+ r.logger.Errorf("send segment to kafka error %v (errors: %d)", err, consecutiveErrors)
+ }
+ continue
+ } else if consecutiveErrors > 0 {
+ consecutiveErrors = 0
+ }
+ }
+}
+
+func (r *kafkaReporter) metricsSendLoop() {
+ consecutiveErrors := 0
+ logFrequency := 30
+ for s := range r.metricsSendCh {
+ payload, err := proto.Marshal(&agentv3.MeterDataCollection{
+ MeterData: s,
+ })
+ if err != nil {
+ r.logger.Errorf("marshal metrics error %v", err)
+ continue
+ }
+ ctx := context.WithValue(context.Background(), internalReporterContextKey, true)
+ err = r.writer.WriteMessages(ctx, kafka.Message{
+ Topic: r.topicMeter,
+ Key: []byte(r.entity.ServiceInstanceName),
+ Value: payload,
+ })
+ if err != nil {
+ consecutiveErrors++
+ if consecutiveErrors == 1 || consecutiveErrors%logFrequency == 0 {
+ r.logger.Errorf("send metrics to kafka error %v (errors: %d)", err, consecutiveErrors)
+ }
+ continue
+ } else if consecutiveErrors > 0 {
+ consecutiveErrors = 0
+ }
+ }
+}
+
+func (r *kafkaReporter) logSendLoop() {
+ consecutiveErrors := 0
+ logFrequency := 30
+ for s := range r.logSendCh {
+ payload, err := proto.Marshal(s)
+ if err != nil {
+ r.logger.Errorf("marshal log error %v", err)
+ continue
+ }
+ ctx := context.WithValue(context.Background(), internalReporterContextKey, true)
+ err = r.writer.WriteMessages(ctx, kafka.Message{
+ Topic: r.topicLogging,
+ Key: []byte(s.Service),
+ Value: payload,
+ })
+ if err != nil {
+ consecutiveErrors++
+ if consecutiveErrors == 1 || consecutiveErrors%logFrequency == 0 {
+ r.logger.Errorf("send log to kafka error %v (errors: %d)", err, consecutiveErrors)
+ }
+ continue
+ } else if consecutiveErrors > 0 {
+ consecutiveErrors = 0
+ }
+ }
+}
+
+func (r *kafkaReporter) SendTracing(spans []reporter.ReportedSpan) {
+ segmentObject := r.transform.TransformSegmentObject(spans)
+ if segmentObject == nil {
+ return
+ }
+ defer func() {
+ // recover the panic caused by close tracingSendCh
+ if err := recover(); err != nil {
+ r.logger.Errorf("reporter segment err %v", err)
+ }
+ }()
+ select {
+ case r.tracingSendCh <- segmentObject:
+ default:
+ r.logger.Errorf("reach max tracing send buffer")
+ }
+}
+
+func (r *kafkaReporter) SendMetrics(metrics []reporter.ReportedMeter) {
+ meters := r.transform.TransformMeterData(metrics)
+ if meters == nil {
+ return
+ }
+ defer func() {
+ // recover the panic caused by close metricsSendCh
+ if err := recover(); err != nil {
+ r.logger.Errorf("reporter metrics err %v", err)
+ }
+ }()
+ select {
+ case r.metricsSendCh <- meters:
+ default:
+ r.logger.Errorf("reach max metrics send buffer")
+ }
+}
+
+func (r *kafkaReporter) SendLog(log *logv3.LogData) {
+ defer func() {
+ if err := recover(); err != nil {
+ r.logger.Errorf("reporter log err %v", err)
+ }
+ }()
+ select {
+ case r.logSendCh <- log:
+ default:
+ }
+}
+
+func (r *kafkaReporter) check() {
+ if r.checkInterval < 0 || r.writer == nil {
+ return
+ }
+ go func() {
+ // initialDelay
+ time.Sleep(r.checkInterval)
+ instancePropertiesSubmitted := false
+ for {
+ if !instancePropertiesSubmitted {
+ instanceProperties := &managementv3.InstanceProperties{
+ Service: r.entity.ServiceName,
+ ServiceInstance: r.entity.ServiceInstanceName,
+ Properties: r.entity.Props,
+ }
+ payload, err := proto.Marshal(instanceProperties)
+ if err != nil {
+ r.logger.Errorf("marshal instance properties error %v", err)
+ time.Sleep(r.checkInterval)
+ continue
+ }
+ ctx := context.WithValue(context.Background(), internalReporterContextKey, true)
+ err = r.writer.WriteMessages(ctx, kafka.Message{
+ Topic: r.topicManagement,
+ Key: []byte(topicKeyRegister + r.entity.ServiceInstanceName),
+ Value: payload,
+ })
+ if err != nil {
+ r.logger.Errorf("send instance properties to kafka error %v", err)
+ time.Sleep(r.checkInterval)
+ continue
+ }
+ instancePropertiesSubmitted = true
+ }
+
+ ping := &managementv3.InstancePingPkg{
+ Service: r.entity.ServiceName,
+ ServiceInstance: r.entity.ServiceInstanceName,
+ }
+ payload, err := proto.Marshal(ping)
+ if err != nil {
+ r.logger.Errorf("marshal instance ping error %v", err)
+ time.Sleep(r.checkInterval)
+ continue
+ }
+ ctx := context.WithValue(context.Background(), internalReporterContextKey, true)
+ err = r.writer.WriteMessages(ctx, kafka.Message{
+ Topic: r.topicManagement,
+ Key: []byte(r.entity.ServiceInstanceName),
+ Value: payload,
+ })
+ if err != nil {
+ r.logger.Errorf("send instance ping to kafka error %v", err)
+ }
+ time.Sleep(r.checkInterval)
+ }
+ }()
+}
+
+func (r *kafkaReporter) ConnectionStatus() reporter.ConnectionStatus {
+ return r.connectionStatus
+}
+
+func (r *kafkaReporter) Close() {
+ if r.bootFlag {
+ if r.tracingSendCh != nil {
+ close(r.tracingSendCh)
+ }
+ if r.metricsSendCh != nil {
+ close(r.metricsSendCh)
+ }
+ if r.logSendCh != nil {
+ close(r.logSendCh)
+ }
+ if err := r.writer.Close(); err != nil {
+ r.logger.Errorf("close kafka writer failed, err: %v", err)
+ }
+ }
+}
diff --git a/plugins/core/reporter/kafka/kafka_opts_enhance.go b/plugins/core/reporter/kafka/kafka_opts_enhance.go
new file mode 100644
index 0000000..cbd80a2
--- /dev/null
+++ b/plugins/core/reporter/kafka/kafka_opts_enhance.go
@@ -0,0 +1,85 @@
+// 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 kafka
+
+import (
+ "time"
+
+ logv3 "skywalking.apache.org/repo/goapi/collect/logging/v3"
+
+ "github.com/segmentio/kafka-go"
+ agentv3 "skywalking.apache.org/repo/goapi/collect/language/agent/v3"
+)
+
+type ReporterOptionKafka func(r *kafkaReporter)
+
+func WithKafkaTopicSegment(topicSegment string) ReporterOptionKafka {
+ return func(r *kafkaReporter) {
+ r.topicSegment = topicSegment
+ }
+}
+
+func WithKafkaTopicMeter(topicMeter string) ReporterOptionKafka {
+ return func(r *kafkaReporter) {
+ r.topicMeter = topicMeter
+ }
+}
+
+func WithKafkaTopicLogging(topicLogging string) ReporterOptionKafka {
+ return func(r *kafkaReporter) {
+ r.topicLogging = topicLogging
+ }
+}
+
+func WithKafkaTopicManagement(topicManagement string) ReporterOptionKafka {
+ return func(r *kafkaReporter) {
+ r.topicManagement = topicManagement
+ }
+}
+
+func WithKafkaMaxSendQueueSize(kafkaMaxSendQueueSize int) ReporterOptionKafka {
+ return func(r *kafkaReporter) {
+ r.tracingSendCh = make(chan *agentv3.SegmentObject, kafkaMaxSendQueueSize)
+ r.metricsSendCh = make(chan []*agentv3.MeterData, kafkaMaxSendQueueSize)
+ r.logSendCh = make(chan *logv3.LogData, kafkaMaxSendQueueSize)
+ }
+}
+
+func WithKafkaBatchSize(batchSize int) ReporterOptionKafka {
+ return func(r *kafkaReporter) {
+ r.writer.BatchSize = batchSize
+ }
+}
+
+func WithKafkaBatchBytes(batchBytes int64) ReporterOptionKafka {
+ return func(r *kafkaReporter) {
+ r.writer.BatchBytes = batchBytes
+ }
+}
+
+func WithKafkaBatchTimeoutMillis(batchTimeoutMillis int) ReporterOptionKafka {
+ return func(r *kafkaReporter) {
+ r.writer.BatchTimeout = time.Duration(batchTimeoutMillis) * time.Millisecond
+ }
+}
+
+func WithKafkaAcks(acks int) ReporterOptionKafka {
+ return func(r *kafkaReporter) {
+ r.writer.RequiredAcks = kafka.RequiredAcks(acks)
+ }
+}
diff --git a/plugins/core/reporter/transform.go b/plugins/core/reporter/transform.go
new file mode 100644
index 0000000..62bca9c
--- /dev/null
+++ b/plugins/core/reporter/transform.go
@@ -0,0 +1,151 @@
+// 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 reporter
+
+import (
+ "time"
+
+ agentv3 "skywalking.apache.org/repo/goapi/collect/language/agent/v3"
+)
+
+type Transform struct {
+ entity *Entity
+}
+
+func NewTransform(entity *Entity) *Transform {
+ return &Transform{
+ entity: entity,
+ }
+}
+
+func (r *Transform) TransformSegmentObject(spans []ReportedSpan) *agentv3.SegmentObject {
+ spanSize := len(spans)
+ if spanSize < 1 {
+ return nil
+ }
+ rootSpan := spans[spanSize-1]
+ rootCtx := rootSpan.Context()
+ segmentObject := &agentv3.SegmentObject{
+ TraceId: rootCtx.GetTraceID(),
+ TraceSegmentId: rootCtx.GetSegmentID(),
+ Spans: make([]*agentv3.SpanObject, spanSize),
+ Service: r.entity.ServiceName,
+ ServiceInstance: r.entity.ServiceInstanceName,
+ }
+ for i, s := range spans {
+ spanCtx := s.Context()
+ segmentObject.Spans[i] = &agentv3.SpanObject{
+ SpanId: spanCtx.GetSpanID(),
+ ParentSpanId: spanCtx.GetParentSpanID(),
+ StartTime: s.StartTime(),
+ EndTime: s.EndTime(),
+ OperationName: s.OperationName(),
+ Peer: s.Peer(),
+ SpanType: s.SpanType(),
+ SpanLayer: s.SpanLayer(),
+ ComponentId: s.ComponentID(),
+ IsError: s.IsError(),
+ Tags: s.Tags(),
+ Logs: s.Logs(),
+ }
+ srr := make([]*agentv3.SegmentReference, 0)
+ if i == (spanSize-1) && spanCtx.GetParentSpanID() > -1 {
+ srr = append(srr, &agentv3.SegmentReference{
+ RefType: agentv3.RefType_CrossThread,
+ TraceId: spanCtx.GetTraceID(),
+ ParentTraceSegmentId: spanCtx.GetParentSegmentID(),
+ ParentSpanId: spanCtx.GetParentSpanID(),
+ ParentService: r.entity.ServiceName,
+ ParentServiceInstance: r.entity.ServiceInstanceName,
+ })
+ }
+ if len(s.Refs()) > 0 {
+ for _, tc := range s.Refs() {
+ srr = append(srr, &agentv3.SegmentReference{
+ RefType: agentv3.RefType_CrossProcess,
+ TraceId: spanCtx.GetTraceID(),
+ ParentTraceSegmentId: tc.GetParentSegmentID(),
+ ParentSpanId: tc.GetParentSpanID(),
+ ParentService: tc.GetParentService(),
+ ParentServiceInstance: tc.GetParentServiceInstance(),
+ ParentEndpoint: tc.GetParentEndpoint(),
+ NetworkAddressUsedAtPeer: tc.GetAddressUsedAtClient(),
+ })
+ }
+ }
+ segmentObject.Spans[i].Refs = srr
+ }
+ return segmentObject
+}
+
+func (r *Transform) TransformMeterData(metrics []ReportedMeter) []*agentv3.MeterData {
+ if len(metrics) == 0 {
+ return nil
+ }
+ meters := make([]*agentv3.MeterData, len(metrics))
+ for i, m := range metrics {
+ meter := &agentv3.MeterData{}
+ switch data := m.(type) {
+ case ReportedMeterSingleValue:
+ meter.Metric = &agentv3.MeterData_SingleValue{
+ SingleValue: &agentv3.MeterSingleValue{
+ Name: data.Name(),
+ Labels: r.transformLabels(data.Labels()),
+ Value: data.Value(),
+ },
+ }
+ case ReportedMeterHistogram:
+ buckets := make([]*agentv3.MeterBucketValue, len(data.BucketValues()))
+ for i, b := range data.BucketValues() {
+ buckets[i] = &agentv3.MeterBucketValue{
+ Bucket: b.Bucket(),
+ Count: b.Count(),
+ IsNegativeInfinity: b.IsNegativeInfinity(),
+ }
+ }
+ meter.Metric = &agentv3.MeterData_Histogram{
+ Histogram: &agentv3.MeterHistogram{
+ Name: data.Name(),
+ Labels: r.transformLabels(data.Labels()),
+ Values: buckets,
+ },
+ }
+ }
+
+ meters[i] = meter
+ }
+
+ meters[0].Service = r.entity.ServiceName
+ meters[0].ServiceInstance = r.entity.ServiceInstanceName
+ meters[0].Timestamp = time.Now().UnixNano() / int64(time.Millisecond)
+ return meters
+}
+
+func (r *Transform) transformLabels(labels map[string]string) []*agentv3.Label {
+ if len(labels) == 0 {
+ return nil
+ }
+ ls := make([]*agentv3.Label, 0)
+ for k, v := range labels {
+ ls = append(ls, &agentv3.Label{
+ Name: k,
+ Value: v,
+ })
+ }
+ return ls
+}
diff --git a/plugins/segmentio-kafka/instrument.go b/plugins/segmentio-kafka/instrument.go
index a7ff008..e2935fd 100644
--- a/plugins/segmentio-kafka/instrument.go
+++ b/plugins/segmentio-kafka/instrument.go
@@ -35,7 +35,7 @@ func NewInstrument() *Instrument {
}
func (i *Instrument) Name() string {
- return "segmentio_kafka"
+ return "segmentio-kafka"
}
func (i *Instrument) BasePackage() string {
diff --git a/plugins/segmentio-kafka/writer_interceptor.go b/plugins/segmentio-kafka/writer_interceptor.go
index 78d93d9..cfd8ef5 100644
--- a/plugins/segmentio-kafka/writer_interceptor.go
+++ b/plugins/segmentio-kafka/writer_interceptor.go
@@ -18,6 +18,8 @@
package segmentiokafka
import (
+ "context"
+
"github.com/segmentio/kafka-go"
"github.com/apache/skywalking-go/plugins/core/operator"
@@ -30,12 +32,18 @@ const (
kafkaWriterComponentID = 40
)
+var internalReporterContextKey = context.Background()
+
type WriterInterceptor struct {
}
func (w *WriterInterceptor) BeforeInvoke(invocation operator.Invocation) error {
writer := invocation.CallerInstance().(*kafka.Writer)
addr, topic := writer.Addr.String(), writer.Topic
+ ctx := invocation.Args()[0].(context.Context)
+ if internal, ok := ctx.Value(internalReporterContextKey).(bool); ok && internal {
+ return nil
+ }
messageList := invocation.Args()[1].([]kafka.Message)
operationName := kafkaWriterPrefix + topic + kafkaWriterSuffix
diff --git a/test/e2e/Makefile b/test/e2e/Makefile
new file mode 100644
index 0000000..1ec2691
--- /dev/null
+++ b/test/e2e/Makefile
@@ -0,0 +1,39 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+GO = go
+GO_BUILD = $(GO) build
+
+GIT_VERSION := $(shell git rev-parse --short HEAD)
+ifeq ($(strip $(GIT_VERSION)),)
+ GIT_VERSION = $(shell grep gitCommit $(VERSION_FILE) | awk -F ': ' '{print $$2}')
+endif
+VERSION ?= $(GIT_VERSION)
+
+os ?= $(OS)
+
+.PHONY: build
+build:
+ mkdir -p dist
+ # go agent
+ @if [ "$(os)" = "windows" ]; then \
+ "$(MAKE)" -C ../../tools/go-agent windows; \
+ cp ../../bin/skywalking-go-agent-$(VERSION)-windows-$(shell go env GOARCH) dist/skywalking-go-agent.exe; \
+ else \
+ "$(MAKE)" -C ../../tools/go-agent linux; \
+ cp ../../bin/skywalking-go-agent-$(VERSION)-linux-$(shell go env GOARCH) dist/skywalking-go-agent; \
+ fi
diff --git a/test/e2e/base/consumer/Dockerfile b/test/e2e/base/consumer/Dockerfile
new file mode 100644
index 0000000..f5158fc
--- /dev/null
+++ b/test/e2e/base/consumer/Dockerfile
@@ -0,0 +1,40 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+FROM golang:1.19 as builder
+
+ENV CGO_ENABLED=0
+ENV GO111MODULE=on
+
+WORKDIR /build
+
+COPY . /build/skywalking-go
+COPY test/e2e/base/consumer /build/consumer
+COPY test/e2e/dist/skywalking-go-agent /build/skywalking-go-agent
+RUN chmod +x /build/skywalking-go-agent
+
+WORKDIR /build/consumer
+RUN go mod edit -replace=github.com/apache/skywalking-go=../skywalking-go
+RUN go mod edit -replace=github.com/apache/skywalking-go/toolkit=../skywalking-go/toolkit
+
+RUN go mod tidy && go build -toolexec="/build/skywalking-go-agent" -a -o consumer
+
+FROM alpine:3.11
+
+WORKDIR /app
+
+COPY --from=builder /build/consumer/consumer /consumer
+
+CMD ["/consumer"]
diff --git a/test/e2e/base/consumer/go.mod b/test/e2e/base/consumer/go.mod
new file mode 100644
index 0000000..5925bd0
--- /dev/null
+++ b/test/e2e/base/consumer/go.mod
@@ -0,0 +1,3 @@
+module test/e2e/base/consumer
+
+go 1.19
\ No newline at end of file
diff --git a/test/e2e/base/consumer/main.go b/test/e2e/base/consumer/main.go
new file mode 100644
index 0000000..9d0b4ad
--- /dev/null
+++ b/test/e2e/base/consumer/main.go
@@ -0,0 +1,133 @@
+// Licensed to the Apache Software Foundation (ASF) under one or more
+// contributor license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright ownership.
+// The ASF licenses this file to You under the Apache License, Version 2.0
+// (the "License"); you may not use this file except in compliance with
+// the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package main
+
+import (
+ "bytes"
+ "fmt"
+ "io"
+ "log"
+ "math/rand"
+ "net/http"
+ "os"
+ "time"
+
+ _ "github.com/apache/skywalking-go"
+ "github.com/apache/skywalking-go/toolkit/logging"
+ "github.com/apache/skywalking-go/toolkit/trace"
+)
+
+var providerURL string
+
+func infoHandler(w http.ResponseWriter, r *http.Request) {
+ log.Printf("Received %s request to /info", r.Method)
+ sleepTime := rand.Intn(500) + 500
+ time.Sleep(time.Duration(sleepTime) * time.Millisecond)
+
+ var resp *http.Response
+ var err error
+
+ resp, err = http.Post(providerURL+"/info", "application/json", bytes.NewBufferString(""))
+ if err != nil {
+ http.Error(w, fmt.Sprintf("Error calling provider: %v", err), http.StatusInternalServerError)
+ return
+ }
+ defer resp.Body.Close()
+
+ body, err := io.ReadAll(resp.Body)
+ if err != nil {
+ http.Error(w, fmt.Sprintf("Error reading provider response: %v", err), http.StatusInternalServerError)
+ return
+ }
+ w.Header().Set("Content-Type", "application/json")
+ _, _ = w.Write(body)
+
+ logging.Debug("this is debug msg", "foo1", "bar1")
+ logging.Info("this is info msg", "foo2", "bar2")
+ logging.Warn("this is warn msg", "foo3", "bar3")
+ logging.Error("this is error msg", "foo4", "bar4")
+ log.Printf("Consumer processed %s request to /info", r.Method)
+}
+
+func usersHandler(w http.ResponseWriter, r *http.Request) {
+ log.Printf("Received %s request to /users", r.Method)
+ sleepTime := rand.Intn(500) + 500
+ time.Sleep(time.Duration(sleepTime) * time.Millisecond)
+
+ var resp *http.Response
+ var err error
+
+ resp, err = http.Post(providerURL+"/users", "application/json", bytes.NewBufferString("{}"))
+ if err != nil {
+ http.Error(w, fmt.Sprintf("Error calling provider: %v", err), http.StatusInternalServerError)
+ return
+ }
+ defer resp.Body.Close()
+
+ body, err := io.ReadAll(resp.Body)
+ if err != nil {
+ http.Error(w, fmt.Sprintf("Error reading provider response: %v", err), http.StatusInternalServerError)
+ return
+ }
+ w.Header().Set("Content-Type", "application/json")
+ _, _ = w.Write(body)
+ log.Printf("Consumer processed %s request to /users", r.Method)
+}
+
+func correlationHandler(w http.ResponseWriter, r *http.Request) {
+ log.Printf("Received %s request to /correlation", r.Method)
+ sleepTime := rand.Intn(500) + 500
+ time.Sleep(time.Duration(sleepTime) * time.Millisecond)
+ trace.SetCorrelation("CONSUMER_KEY", "consumer")
+
+ var resp *http.Response
+ var err error
+
+ resp, err = http.Post(providerURL+"/correlation", "application/json", bytes.NewBufferString(""))
+ if err != nil {
+ http.Error(w, fmt.Sprintf("Error calling provider: %v", err), http.StatusInternalServerError)
+ return
+ }
+ defer resp.Body.Close()
+
+ body, err := io.ReadAll(resp.Body)
+ if err != nil {
+ http.Error(w, fmt.Sprintf("Error reading provider response: %v", err), http.StatusInternalServerError)
+ return
+ }
+ w.Header().Set("Content-Type", "application/json")
+ _, _ = w.Write(body)
+ log.Printf("Consumer processed %s request to /correlation", r.Method)
+}
+
+func main() {
+ providerURL = os.Getenv("PROVIDER_URL")
+ if providerURL == "" {
+ providerURL = "http://localhost:8080"
+ }
+ log.Printf("Provider URL: %s\n", providerURL)
+
+ http.HandleFunc("/health", func(w http.ResponseWriter, r *http.Request) {
+ _, _ = w.Write([]byte("ok"))
+ })
+ http.HandleFunc("/info", infoHandler)
+ http.HandleFunc("/users", usersHandler)
+ http.HandleFunc("/correlation", correlationHandler)
+
+ if err := http.ListenAndServe(":8080", nil); err != nil {
+ log.Fatalf("Error starting server: %v", err)
+ }
+}
diff --git a/test/e2e/base/docker-compose.yml b/test/e2e/base/docker-compose.yml
new file mode 100644
index 0000000..39a5dcc
--- /dev/null
+++ b/test/e2e/base/docker-compose.yml
@@ -0,0 +1,56 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+version: "2.1"
+
+services:
+ provider:
+ build:
+ context: ../../..
+ dockerfile: test/e2e/base/provider/Dockerfile
+ expose:
+ - 8080
+ networks:
+ - e2e
+ environment:
+ SW_AGENT_NAME: go-provider
+ SW_AGENT_REPORTER_GRPC_BACKEND_SERVICE: oap:11800
+ SW_AGENT_INSTANCE_NAME: provider1
+ healthcheck:
+ test: ["CMD", "wget", "-q", "-O", "-", "http://127.0.0.1:8080/health"]
+ interval: 5s
+ timeout: 60s
+ retries: 120
+
+ consumer:
+ build:
+ context: ../../..
+ dockerfile: test/e2e/base/consumer/Dockerfile
+ expose:
+ - 8080
+ networks:
+ - e2e
+ environment:
+ SW_AGENT_NAME: go-consumer
+ SW_AGENT_REPORTER_GRPC_BACKEND_SERVICE: oap:11800
+ SW_AGENT_INSTANCE_NAME: consumer1
+ healthcheck:
+ test: ["CMD", "wget", "-q", "-O", "-", "http://127.0.0.1:8080/health"]
+ interval: 5s
+ timeout: 60s
+ retries: 120
+
+networks:
+ e2e:
diff --git a/test/e2e/base/provider/Dockerfile b/test/e2e/base/provider/Dockerfile
new file mode 100644
index 0000000..7b6a776
--- /dev/null
+++ b/test/e2e/base/provider/Dockerfile
@@ -0,0 +1,40 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+FROM golang:1.19 as builder
+
+ENV CGO_ENABLED=0
+ENV GO111MODULE=on
+
+WORKDIR /build
+
+COPY . /build/skywalking-go
+COPY test/e2e/base/provider /build/provider
+COPY test/e2e/dist/skywalking-go-agent /build/skywalking-go-agent
+RUN chmod +x /build/skywalking-go-agent
+
+WORKDIR /build/provider
+RUN go mod edit -replace=github.com/apache/skywalking-go=../skywalking-go
+RUN go mod edit -replace=github.com/apache/skywalking-go/toolkit=../skywalking-go/toolkit
+
+RUN go mod tidy && go build -toolexec="/build/skywalking-go-agent" -a -o provider
+
+FROM alpine:3.11
+
+WORKDIR /app
+
+COPY --from=builder /build/provider/provider /provider
+
+CMD ["/provider"]
diff --git a/test/e2e/base/provider/go.mod b/test/e2e/base/provider/go.mod
new file mode 100644
index 0000000..57d0d7e
--- /dev/null
+++ b/test/e2e/base/provider/go.mod
@@ -0,0 +1,3 @@
+module test/e2e/base/provider
+
+go 1.19
\ No newline at end of file
diff --git a/test/e2e/base/provider/main.go b/test/e2e/base/provider/main.go
new file mode 100644
index 0000000..e7e6b4b
--- /dev/null
+++ b/test/e2e/base/provider/main.go
@@ -0,0 +1,83 @@
+// Licensed to the Apache Software Foundation (ASF) under one or more
+// contributor license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright ownership.
+// The ASF licenses this file to You under the Apache License, Version 2.0
+// (the "License"); you may not use this file except in compliance with
+// the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package main
+
+import (
+ "encoding/json"
+ "fmt"
+ "log"
+ "math/rand"
+ "net/http"
+ "time"
+
+ _ "github.com/apache/skywalking-go"
+ "github.com/apache/skywalking-go/toolkit/trace"
+)
+
+type User struct {
+ ID int `json:"id"`
+ Name string `json:"name"`
+}
+
+func main() {
+ http.HandleFunc("/health", func(w http.ResponseWriter, r *http.Request) {
+ _, _ = w.Write([]byte("ok"))
+ })
+
+ http.HandleFunc("/info", func(w http.ResponseWriter, r *http.Request) {
+ log.Printf("Received %s request to /info", r.Method)
+
+ sleepTime := rand.Intn(500) + 500
+ time.Sleep(time.Duration(sleepTime) * time.Millisecond)
+
+ _, _ = w.Write([]byte("This is provider info"))
+ log.Printf("Provider processed %s request to /info", r.Method)
+ })
+
+ http.HandleFunc("/users", func(w http.ResponseWriter, r *http.Request) {
+ log.Printf("Received %s request to /users", r.Method)
+ sleepTime := rand.Intn(500) + 500
+ time.Sleep(time.Duration(sleepTime) * time.Millisecond)
+ user := User{
+ ID: 1,
+ Name: "skywalking-go",
+ }
+ jsonData, err := json.Marshal(user)
+ if err != nil {
+ http.Error(w, fmt.Sprintf("Error encoding JSON: %v", err), http.StatusInternalServerError)
+ return
+ }
+
+ w.Header().Set("Content-Type", "application/json")
+ _, _ = w.Write(jsonData)
+ log.Printf("Provider processed %s request to /users", r.Method)
+ })
+
+ http.HandleFunc("/correlation", func(w http.ResponseWriter, r *http.Request) {
+ log.Printf("Received %s request to /correlation", r.Method)
+ sleepTime := rand.Intn(500) + 500
+ time.Sleep(time.Duration(sleepTime) * time.Millisecond)
+ trace.SetCorrelation("PROVIDER_KEY", "provider")
+
+ correlation := trace.GetCorrelation("PROVIDER_KEY") + "_" + trace.GetCorrelation("MIDDLE_KEY") + "_" + trace.GetCorrelation("PROVIDER_KEY")
+ _, _ = w.Write([]byte(correlation))
+ log.Printf("Provider processed %s request to /correlation", r.Method)
+ })
+
+ if err := http.ListenAndServe(":8080", nil); err != nil {
+ log.Fatalf("Error starting server: %v", err)
+ }
+}
diff --git a/test/e2e/case/expected/logs-list.yml b/test/e2e/case/expected/logs-list.yml
new file mode 100644
index 0000000..5c4aa56
--- /dev/null
+++ b/test/e2e/case/expected/logs-list.yml
@@ -0,0 +1,88 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+logs:
+ {{- contains .logs }}
+ - servicename: go-consumer
+ serviceid: {{ b64enc "go-consumer" }}.1
+ serviceinstancename: consumer1
+ serviceinstanceid: {{ b64enc "go-consumer" }}.1_{{ b64enc "consumer1" }}
+ endpointname: POST:/info
+ endpointid: {{ b64enc "go-consumer" }}.1_{{ b64enc "POST:/info" }}
+ traceid: {{ notEmpty .traceid }}
+ timestamp: {{ gt .timestamp 0 }}
+ contenttype: TEXT
+ content: this is debug msg
+ tags:
+ {{- contains .tags }}
+ - key: LEVEL
+ value: debug
+ - key: foo1
+ value: bar1
+ {{- end }}
+ - servicename: go-consumer
+ serviceid: {{ b64enc "go-consumer" }}.1
+ serviceinstancename: consumer1
+ serviceinstanceid: {{ b64enc "go-consumer" }}.1_{{ b64enc "consumer1" }}
+ endpointname: POST:/info
+ endpointid: {{ b64enc "go-consumer" }}.1_{{ b64enc "POST:/info" }}
+ traceid: {{ notEmpty .traceid }}
+ timestamp: {{ gt .timestamp 0 }}
+ contenttype: TEXT
+ content: this is info msg
+ tags:
+ {{- contains .tags }}
+ - key: LEVEL
+ value: info
+ - key: foo2
+ value: bar2
+ {{- end }}
+ - servicename: go-consumer
+ serviceid: {{ b64enc "go-consumer" }}.1
+ serviceinstancename: consumer1
+ serviceinstanceid: {{ b64enc "go-consumer" }}.1_{{ b64enc "consumer1" }}
+ endpointname: POST:/info
+ endpointid: {{ b64enc "go-consumer" }}.1_{{ b64enc "POST:/info" }}
+ traceid: {{ notEmpty .traceid }}
+ timestamp: {{ gt .timestamp 0 }}
+ contenttype: TEXT
+ content: this is warn msg
+ tags:
+ {{- contains .tags }}
+ - key: LEVEL
+ value: warn
+ - key: foo3
+ value: bar3
+ {{- end }}
+ - servicename: go-consumer
+ serviceid: {{ b64enc "go-consumer" }}.1
+ serviceinstancename: consumer1
+ serviceinstanceid: {{ b64enc "go-consumer" }}.1_{{ b64enc "consumer1" }}
+ endpointname: POST:/info
+ endpointid: {{ b64enc "go-consumer" }}.1_{{ b64enc "POST:/info" }}
+ traceid: {{ notEmpty .traceid }}
+ timestamp: {{ gt .timestamp 0 }}
+ contenttype: TEXT
+ content: this is error msg
+ tags:
+ {{- contains .tags }}
+ - key: LEVEL
+ value: error
+ - key: foo4
+ value: bar4
+ {{- end }}
+ {{- end }}
+debuggingtrace: null
+errorreason: null
\ No newline at end of file
diff --git a/test/e2e/case/expected/metrics-has-value.yml b/test/e2e/case/expected/metrics-has-value.yml
new file mode 100644
index 0000000..a33d8a2
--- /dev/null
+++ b/test/e2e/case/expected/metrics-has-value.yml
@@ -0,0 +1,21 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+{{- contains . }}
+- key: {{ notEmpty .key }}
+ value:
+ value: {{ ge .value.value 0 }}
+ isemptyvalue: false
+{{- end }}
\ No newline at end of file
diff --git a/test/e2e/case/expected/service-endpoint-consumer.yml b/test/e2e/case/expected/service-endpoint-consumer.yml
new file mode 100644
index 0000000..44f8762
--- /dev/null
+++ b/test/e2e/case/expected/service-endpoint-consumer.yml
@@ -0,0 +1,19 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+{{- contains .}}
+- id: {{ b64enc "go-consumer" }}.1_{{ b64enc "POST:/info" }}
+ name: POST:/info
+{{- end}}
\ No newline at end of file
diff --git a/test/e2e/case/expected/service-endpoint-provider.yml b/test/e2e/case/expected/service-endpoint-provider.yml
new file mode 100644
index 0000000..b332e0e
--- /dev/null
+++ b/test/e2e/case/expected/service-endpoint-provider.yml
@@ -0,0 +1,19 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+{{- contains .}}
+- id: {{ b64enc "go-provider" }}.1_{{ b64enc "POST:/info" }}
+ name: POST:/info
+{{- end}}
\ No newline at end of file
diff --git a/test/e2e/case/expected/service-instance-consumer.yml b/test/e2e/case/expected/service-instance-consumer.yml
new file mode 100644
index 0000000..f623e63
--- /dev/null
+++ b/test/e2e/case/expected/service-instance-consumer.yml
@@ -0,0 +1,34 @@
+# 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.
+
+{{- contains .}}
+- id: {{ b64enc "go-consumer" }}.1_{{ b64enc "consumer1" }}
+ name: {{ notEmpty .name }}
+ attributes:
+ {{- contains .attributes }}
+ - name: Process No.
+ value: "1"
+ - name: hostname
+ value: {{ notEmpty .value }}
+ - name: OS Name
+ value: linux
+ - name: ipv4s
+ value: {{ notEmpty .value }}
+ {{- end }}
+ language: GO
+ instanceuuid: {{ b64enc "go-consumer" }}.1_{{ b64enc "consumer1" }}
+{{- end}}
\ No newline at end of file
diff --git a/test/e2e/case/expected/service-instance-provider.yml b/test/e2e/case/expected/service-instance-provider.yml
new file mode 100644
index 0000000..5aad3e0
--- /dev/null
+++ b/test/e2e/case/expected/service-instance-provider.yml
@@ -0,0 +1,36 @@
+# 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.
+
+{{- contains .}}
+- id: {{ b64enc "go-provider" }}.1_{{ b64enc "provider1" }}
+ name: {{ notEmpty .name }}
+ attributes:
+ {{- contains .attributes }}
+ - name: Process No.
+ value: "1"
+ - name: hostname
+ value: {{ notEmpty .value }}
+ - name: OS Name
+ value: linux
+ - name: ipv4s
+ value: {{ notEmpty .value }}
+ {{- end }}
+ language: GO
+ instanceuuid: {{ b64enc "go-provider" }}.1_{{ b64enc "provider1" }}
+{{- end}}
+
+
diff --git a/test/e2e/case/expected/service.yml b/test/e2e/case/expected/service.yml
new file mode 100644
index 0000000..25f9e3e
--- /dev/null
+++ b/test/e2e/case/expected/service.yml
@@ -0,0 +1,33 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+{{- contains . }}
+- id: {{ b64enc "go-provider" }}.1
+ name: go-provider
+ group: ""
+ shortname: go-provider
+ normal: true
+ layers:
+ - GENERAL
+ - SO11Y_GO_AGENT
+- id: {{ b64enc "go-consumer" }}.1
+ name: go-consumer
+ group: ""
+ shortname: go-consumer
+ normal: true
+ layers:
+ - GENERAL
+ - SO11Y_GO_AGENT
+{{- end }}
diff --git a/test/e2e/case/expected/trace-info-detail.yml b/test/e2e/case/expected/trace-info-detail.yml
new file mode 100644
index 0000000..024d385
--- /dev/null
+++ b/test/e2e/case/expected/trace-info-detail.yml
@@ -0,0 +1,47 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+spans:
+ {{- contains .spans}}
+ - traceid: {{ notEmpty .traceid }}
+ segmentid: {{ notEmpty .segmentid }}
+ spanid: 0
+ parentspanid: -1
+ refs: []
+ servicecode: go-consumer
+ serviceinstancename: consumer1
+ starttime: {{ gt .starttime 0 }}
+ endtime: {{ gt .endtime 0 }}
+ endpointname: POST:/info
+ type: Entry
+ peer: ""
+ component: GoHttpServer
+ iserror: false
+ layer: Http
+ tags:
+ {{- contains .tags }}
+ - key: http.method
+ value: POST
+ - key: url
+ value: {{ notEmpty .value }}
+ - key: status_code
+ value: '200'
+ {{- end }}
+ logs: []
+ attachedevents: []
+ {{- end }}
+debuggingtrace: null
+
+
diff --git a/test/e2e/case/expected/trace-users-detail.yml b/test/e2e/case/expected/trace-users-detail.yml
new file mode 100644
index 0000000..3a1030e
--- /dev/null
+++ b/test/e2e/case/expected/trace-users-detail.yml
@@ -0,0 +1,43 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+spans:
+ {{- contains .spans}}
+ - traceid: {{ notEmpty .traceid }}
+ segmentid: {{ notEmpty .segmentid }}
+ spanid: 0
+ parentspanid: -1
+ refs: []
+ servicecode: {{ notEmpty .servicecode }}
+ serviceinstancename: {{ notEmpty .serviceinstancename }}
+ starttime: {{ gt .starttime 0 }}
+ endtime: {{ gt .endtime 0 }}
+ endpointname: POST:/users
+ type: Entry
+ peer: ""
+ component: Tomcat
+ iserror: false
+ layer: Http
+ tags:
+ {{- contains .tags }}
+ - key: http.method
+ value: POST
+ - key: url
+ value: {{ notEmpty .value }}
+ - key: http.status_code
+ value: '200'
+ {{- end }}
+ logs: []
+ {{- end }}
diff --git a/test/e2e/case/expected/traces-list-consumer.yml b/test/e2e/case/expected/traces-list-consumer.yml
new file mode 100644
index 0000000..6cc6f54
--- /dev/null
+++ b/test/e2e/case/expected/traces-list-consumer.yml
@@ -0,0 +1,29 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+debuggingtrace: null
+traces:
+ {{- contains .traces }}
+ - segmentid: {{ notEmpty .segmentid }}
+ endpointnames:
+ {{- contains .endpointnames }}
+ - POST:/info
+ {{- end }}
+ duration: {{ ge .duration 0 }}
+ start: "{{ notEmpty .start}}"
+ iserror: false
+ traceids:
+ - {{ (index .traceids 0) }}
+ {{- end }}
diff --git a/test/e2e/case/expected/traces-list-provider.yml b/test/e2e/case/expected/traces-list-provider.yml
new file mode 100644
index 0000000..6cc6f54
--- /dev/null
+++ b/test/e2e/case/expected/traces-list-provider.yml
@@ -0,0 +1,29 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+debuggingtrace: null
+traces:
+ {{- contains .traces }}
+ - segmentid: {{ notEmpty .segmentid }}
+ endpointnames:
+ {{- contains .endpointnames }}
+ - POST:/info
+ {{- end }}
+ duration: {{ ge .duration 0 }}
+ start: "{{ notEmpty .start}}"
+ iserror: false
+ traceids:
+ - {{ (index .traceids 0) }}
+ {{- end }}
diff --git a/test/e2e/case/grpc/docker-compose.yml b/test/e2e/case/grpc/docker-compose.yml
new file mode 100644
index 0000000..587f22e
--- /dev/null
+++ b/test/e2e/case/grpc/docker-compose.yml
@@ -0,0 +1,64 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+version: "2.1"
+
+services:
+ banyandb:
+ extends:
+ file: ../../script/docker-compose/base-compose.yml
+ service: banyandb
+ ports:
+ - 17912
+
+ oap:
+ extends:
+ file: ../../script/docker-compose/base-compose.yml
+ service: oap
+ ports:
+ - 12800
+ depends_on:
+ banyandb:
+ condition: service_healthy
+
+ provider:
+ extends:
+ file: ../../base/docker-compose.yml
+ service: provider
+ environment:
+ SW_AGENT_REPORTER_TYPE: grpc
+ ports:
+ - 8080
+ depends_on:
+ oap:
+ condition: service_healthy
+
+ consumer:
+ extends:
+ file: ../../base/docker-compose.yml
+ service: consumer
+ environment:
+ SW_AGENT_REPORTER_TYPE: grpc
+ PROVIDER_URL: http://provider:8080
+ ports:
+ - 8080
+ depends_on:
+ oap:
+ condition: service_healthy
+ provider:
+ condition: service_healthy
+
+networks:
+ e2e:
diff --git a/test/e2e/case/grpc/e2e.yaml b/test/e2e/case/grpc/e2e.yaml
new file mode 100644
index 0000000..b5e3762
--- /dev/null
+++ b/test/e2e/case/grpc/e2e.yaml
@@ -0,0 +1,88 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+# This file is used to show how to write configuration files and can be used to test.
+
+setup:
+ env: compose
+ file: docker-compose.yml
+ timeout: 1200
+ init-system-environment: ../../script/env
+ steps:
+ - name: set PATH
+ command: export PATH=/tmp/skywalking-infra-e2e/bin:$PATH
+ - name: install yq
+ command: bash test/e2e/script/prepare/setup-e2e/install.sh yq
+ - name: install swctl
+ command: bash test/e2e/script/prepare/setup-e2e/install.sh swctl
+
+trigger:
+ action: http
+ interval: 3s
+ times: 5
+ url: http://${consumer_host}:${consumer_8080}/info
+ method: POST
+
+verify:
+ # verify with retry strategy
+ retry:
+ # max retry count
+ count: 20
+ # the interval between two retries, in millisecond.
+ interval: 10s
+ cases:
+ # service list
+ - query: swctl --display yaml --base-url=http://${oap_host}:${oap_12800}/graphql service ls
+ expected: ../expected/service.yml
+ # service instance list
+ - query: swctl --display yaml --base-url=http://${oap_host}:${oap_12800}/graphql instance list --service-name=go-provider
+ expected: ../expected/service-instance-provider.yml
+ - query: swctl --display yaml --base-url=http://${oap_host}:${oap_12800}/graphql instance list --service-name=go-consumer
+ expected: ../expected/service-instance-consumer.yml
+ # service endpoint
+ - query: swctl --display yaml --base-url=http://${oap_host}:${oap_12800}/graphql endpoint list --keyword=info --service-name=go-provider
+ expected: ../expected/service-endpoint-provider.yml
+ - query: swctl --display yaml --base-url=http://${oap_host}:${oap_12800}/graphql endpoint list --keyword=info --service-name=go-consumer
+ expected: ../expected/service-endpoint-consumer.yml
+ # trace segment list
+ - query: swctl --display yaml --base-url=http://${oap_host}:${oap_12800}/graphql trace ls --service-name=go-provider
+ expected: ../expected/traces-list-provider.yml
+ - query: swctl --display yaml --base-url=http://${oap_host}:${oap_12800}/graphql trace ls --service-name=go-consumer
+ expected: ../expected/traces-list-consumer.yml
+ # trace detail
+ - query: |
+ swctl --display yaml --base-url=http://${oap_host}:${oap_12800}/graphql trace $( \
+ swctl --display yaml --base-url=http://${oap_host}:${oap_12800}/graphql trace ls --service-name=go-consumer\
+ | yq e '.traces | select(.[].endpointnames[0]=="POST:/info") | .[0].traceids[0]' - \
+ )
+ expected: ../expected/trace-info-detail.yml
+ # service metrics
+ - query: swctl --display yaml --base-url=http://${oap_host}:${oap_12800}/graphql metrics linear --name service_sla --service-name=go-provider |yq e 'to_entries' -
+ expected: ../expected/metrics-has-value.yml
+ - query: swctl --display yaml --base-url=http://${oap_host}:${oap_12800}/graphql metrics linear --name service_sla --service-name=go-consumer |yq e 'to_entries' -
+ expected: ../expected/metrics-has-value.yml
+ # service endpoint metrics
+ - query: swctl --display yaml --base-url=http://${oap_host}:${oap_12800}/graphql metrics linear --name endpoint_cpm --endpoint-name=POST:/info --service-name=go-provider |yq e 'to_entries' -
+ expected: ../expected/metrics-has-value.yml
+ - query: swctl --display yaml --base-url=http://${oap_host}:${oap_12800}/graphql metrics linear --name endpoint_cpm --endpoint-name=POST:/info --service-name=go-provider |yq e 'to_entries' -
+ expected: ../expected/metrics-has-value.yml
+
+ # logs list
+ - query: |
+ swctl --display yaml --base-url=http://${oap_host}:${oap_12800}/graphql logs list --service-name=go-consumer --trace-id=$( \
+ swctl --display yaml --base-url=http://${oap_host}:${oap_12800}/graphql trace ls \
+ | yq e '.traces | select(.[].endpointnames[0]=="POST:/info") | .[0].traceids[0]' -
+ )
+ expected: ../expected/logs-list.yml
diff --git a/test/e2e/case/kafka/docker-compose.yml b/test/e2e/case/kafka/docker-compose.yml
new file mode 100644
index 0000000..1679bf4
--- /dev/null
+++ b/test/e2e/case/kafka/docker-compose.yml
@@ -0,0 +1,141 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+version: "2.1"
+
+services:
+ zookeeper:
+ image: zookeeper:3.9.2
+ hostname: zookeeper
+ expose:
+ - 2181
+ networks:
+ - e2e
+ environment:
+ - ALLOW_ANONYMOUS_LOGIN=yes
+ healthcheck:
+ test: ["CMD", "sh", "-c", "nc -nz 127.0.0.1 2181"]
+ interval: 5s
+ timeout: 60s
+ retries: 120
+
+ broker-a:
+ image: bitnami/kafka:2.4.1
+ hostname: broker-a
+ expose:
+ - 9092
+ networks:
+ - e2e
+ environment:
+ - KAFKA_ZOOKEEPER_CONNECT=zookeeper:2181
+ - KAFKA_BROKER_ID=10
+ - ALLOW_PLAINTEXT_LISTENER=yes
+ depends_on:
+ zookeeper:
+ condition: service_healthy
+ healthcheck:
+ test:
+ ["CMD", "kafka-topics.sh", "--list", "--zookeeper", "zookeeper:2181"]
+ interval: 5s
+ timeout: 60s
+ retries: 120
+
+ broker-b:
+ image: bitnami/kafka:2.4.1
+ hostname: broker-b
+ expose:
+ - 9092
+ networks:
+ - e2e
+ environment:
+ - KAFKA_ZOOKEEPER_CONNECT=zookeeper:2181
+ - KAFKA_BROKER_ID=24
+ - ALLOW_PLAINTEXT_LISTENER=yes
+ depends_on:
+ zookeeper:
+ condition: service_healthy
+ healthcheck:
+ test:
+ ["CMD", "kafka-topics.sh", "--list", "--zookeeper", "zookeeper:2181"]
+ interval: 5s
+ timeout: 60s
+ retries: 12
+
+ banyandb:
+ extends:
+ file: ../../script/docker-compose/base-compose.yml
+ service: banyandb
+ ports:
+ - 17912
+
+ oap:
+ extends:
+ file: ../../script/docker-compose/base-compose.yml
+ service: oap
+ environment:
+ - SW_KAFKA_FETCHER=default
+ - SW_KAFKA_FETCHER_ENABLE_NATIVE_PROTO_LOG=true
+ - SW_KAFKA_FETCHER_SERVERS=broker-a:9092,broker-b:9092
+ - SW_KAFKA_FETCHER_PARTITIONS=2
+ - SW_KAFKA_FETCHER_PARTITIONS_FACTOR=1
+ ports:
+ - 12800
+ depends_on:
+ banyandb:
+ condition: service_healthy
+ broker-a:
+ condition: service_healthy
+ broker-b:
+ condition: service_healthy
+
+ kafkaprovider:
+ extends:
+ file: ../../base/docker-compose.yml
+ service: provider
+ environment:
+ - SW_AGENT_REPORTER_TYPE=kafka
+ - SW_AGENT_REPORTER_KAFKA_BROKERS=broker-a:9092,broker-b:9092
+ ports:
+ - 8080
+ depends_on:
+ oap:
+ condition: service_healthy
+ broker-a:
+ condition: service_healthy
+ broker-b:
+ condition: service_healthy
+
+ kafkaconsumer:
+ extends:
+ file: ../../base/docker-compose.yml
+ service: consumer
+ environment:
+ - SW_AGENT_REPORTER_TYPE=kafka
+ - SW_AGENT_REPORTER_KAFKA_BROKERS=broker-a:9092,broker-b:9092
+ - PROVIDER_URL=http://kafkaprovider:8080
+ ports:
+ - 8080
+ depends_on:
+ oap:
+ condition: service_healthy
+ kafkaprovider:
+ condition: service_healthy
+ broker-a:
+ condition: service_healthy
+ broker-b:
+ condition: service_healthy
+
+networks:
+ e2e:
diff --git a/test/e2e/case/kafka/e2e.yaml b/test/e2e/case/kafka/e2e.yaml
new file mode 100644
index 0000000..c466938
--- /dev/null
+++ b/test/e2e/case/kafka/e2e.yaml
@@ -0,0 +1,88 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+# This file is used to show how to write configuration files and can be used to test.
+
+setup:
+ env: compose
+ file: docker-compose.yml
+ timeout: 1200
+ init-system-environment: ../../script/env
+ steps:
+ - name: set PATH
+ command: export PATH=/tmp/skywalking-infra-e2e/bin:$PATH
+ - name: install yq
+ command: bash test/e2e/script/prepare/setup-e2e/install.sh yq
+ - name: install swctl
+ command: bash test/e2e/script/prepare/setup-e2e/install.sh swctl
+
+trigger:
+ action: http
+ interval: 3s
+ times: 5
+ url: http://${kafkaconsumer_host}:${kafkaconsumer_8080}/info
+ method: POST
+
+verify:
+ # verify with retry strategy
+ retry:
+ # max retry count
+ count: 20
+ # the interval between two retries, in millisecond.
+ interval: 10s
+ cases:
+ # service list
+ - query: swctl --display yaml --base-url=http://${oap_host}:${oap_12800}/graphql service ls
+ expected: ../expected/service.yml
+ # service instance list
+ - query: swctl --display yaml --base-url=http://${oap_host}:${oap_12800}/graphql instance list --service-name=go-provider
+ expected: ../expected/service-instance-provider.yml
+ - query: swctl --display yaml --base-url=http://${oap_host}:${oap_12800}/graphql instance list --service-name=go-consumer
+ expected: ../expected/service-instance-consumer.yml
+ # service endpoint
+ - query: swctl --display yaml --base-url=http://${oap_host}:${oap_12800}/graphql endpoint list --keyword=info --service-name=go-provider
+ expected: ../expected/service-endpoint-provider.yml
+ - query: swctl --display yaml --base-url=http://${oap_host}:${oap_12800}/graphql endpoint list --keyword=info --service-name=go-consumer
+ expected: ../expected/service-endpoint-consumer.yml
+ # trace segment list
+ - query: swctl --display yaml --base-url=http://${oap_host}:${oap_12800}/graphql trace ls --service-name=go-provider
+ expected: ../expected/traces-list-provider.yml
+ - query: swctl --display yaml --base-url=http://${oap_host}:${oap_12800}/graphql trace ls --service-name=go-consumer
+ expected: ../expected/traces-list-consumer.yml
+ # trace detail
+ - query: |
+ swctl --display yaml --base-url=http://${oap_host}:${oap_12800}/graphql trace $( \
+ swctl --display yaml --base-url=http://${oap_host}:${oap_12800}/graphql trace ls --service-name=go-consumer\
+ | yq e '.traces | select(.[].endpointnames[0]=="POST:/info") | .[0].traceids[0]' - \
+ )
+ expected: ../expected/trace-info-detail.yml
+ # service metrics
+ - query: swctl --display yaml --base-url=http://${oap_host}:${oap_12800}/graphql metrics linear --name service_sla --service-name=go-provider |yq e 'to_entries' -
+ expected: ../expected/metrics-has-value.yml
+ - query: swctl --display yaml --base-url=http://${oap_host}:${oap_12800}/graphql metrics linear --name service_sla --service-name=go-consumer |yq e 'to_entries' -
+ expected: ../expected/metrics-has-value.yml
+ # service endpoint metrics
+ - query: swctl --display yaml --base-url=http://${oap_host}:${oap_12800}/graphql metrics linear --name endpoint_cpm --endpoint-name=POST:/info --service-name=go-provider |yq e 'to_entries' -
+ expected: ../expected/metrics-has-value.yml
+ - query: swctl --display yaml --base-url=http://${oap_host}:${oap_12800}/graphql metrics linear --name endpoint_cpm --endpoint-name=POST:/info --service-name=go-provider |yq e 'to_entries' -
+ expected: ../expected/metrics-has-value.yml
+
+ # logs list
+ - query: |
+ swctl --display yaml --base-url=http://${oap_host}:${oap_12800}/graphql logs list --service-name=go-consumer --trace-id=$( \
+ swctl --display yaml --base-url=http://${oap_host}:${oap_12800}/graphql trace ls \
+ | yq e '.traces | select(.[].endpointnames[0]=="POST:/info") | .[0].traceids[0]' -
+ )
+ expected: ../expected/logs-list.yml
diff --git a/test/e2e/script/docker-compose/base-compose.yml b/test/e2e/script/docker-compose/base-compose.yml
new file mode 100644
index 0000000..814a61a
--- /dev/null
+++ b/test/e2e/script/docker-compose/base-compose.yml
@@ -0,0 +1,61 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+version: "2.1"
+
+services:
+ oap:
+ image: ghcr.io/apache/skywalking/oap:${SW_OAP_COMMIT}
+ expose:
+ - 11800
+ - 12800
+ - 10051
+ - 5005
+ networks:
+ - e2e
+ volumes:
+ - ./../prepare/setup-oap/log4j2.xml:/skywalking/config/log4j2.xml
+ environment:
+ SW_STORAGE_BANYANDB_TARGETS: banyandb:17912
+ healthcheck:
+ test: ["CMD", "bash", "-c", "cat < /dev/null > /dev/tcp/127.0.0.1/11800"]
+ interval: 5s
+ timeout: 60s
+ retries: 120
+
+ ui:
+ image: ghcr.io/apache/skywalking/ui:${SW_UI_COMMIT}
+ expose:
+ - 8080
+ networks:
+ - e2e
+ environment:
+ - SW_OAP_ADDRESS=http://oap:12800
+
+ banyandb:
+ image: "ghcr.io/apache/skywalking-banyandb:${SW_BANYANDB_COMMIT}"
+ networks:
+ - e2e
+ expose:
+ - 17912
+ command: standalone --stream-root-path /tmp/stream-data --measure-root-path /tmp/measure-data
+ healthcheck:
+ test: ["CMD", "sh", "-c", "nc -nz 127.0.0.1 17912"]
+ interval: 5s
+ timeout: 60s
+ retries: 120
+
+networks:
+ e2e:
diff --git a/test/e2e/script/env b/test/e2e/script/env
new file mode 100644
index 0000000..d2d63a0
--- /dev/null
+++ b/test/e2e/script/env
@@ -0,0 +1,19 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+SW_CTL_COMMIT=67cbc89dd7b214d5791321a7ca992f940cb586ba
+SW_OAP_COMMIT=b55fa2a5193618cdd7d699ad96fa20ceca9d507e
+SW_UI_COMMIT=b55fa2a5193618cdd7d699ad96fa20ceca9d507e
+SW_BANYANDB_COMMIT=458041a561b0acc1f2ed37690df2ce753b791283
\ No newline at end of file
diff --git a/test/e2e/script/prepare/setup-e2e/install-swctl.sh b/test/e2e/script/prepare/setup-e2e/install-swctl.sh
new file mode 100755
index 0000000..f9bcf34
--- /dev/null
+++ b/test/e2e/script/prepare/setup-e2e/install-swctl.sh
@@ -0,0 +1,39 @@
+#!/usr/bin/env bash
+
+# ----------------------------------------------------------------------------
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+# ----------------------------------------------------------------------------
+
+BASE_DIR=$1
+BIN_DIR=$2
+
+install_swctl() {
+ mkdir -p $BASE_DIR/swctl && cd $BASE_DIR/swctl
+ curl -kLo skywalking-cli.tar.gz https://github.com/apache/skywalking-cli/archive/${SW_CTL_COMMIT}.tar.gz
+ tar -zxf skywalking-cli.tar.gz --strip=1
+ VERSION=${SW_CTL_COMMIT} make install DESTDIR=$BIN_DIR
+}
+
+if ! command -v swctl &> /dev/null; then
+ echo "swctl is not installed"
+ install_swctl
+elif ! swctl --version | grep -q "${SW_CTL_COMMIT}"; then
+ # Check if the installed version is correct
+ echo "swctl is already installed, but version is not ${SW_CTL_COMMIT}, will re-install it"
+ install_swctl
+fi
diff --git a/test/e2e/script/prepare/setup-e2e/install-yq.sh b/test/e2e/script/prepare/setup-e2e/install-yq.sh
new file mode 100755
index 0000000..e0cca70
--- /dev/null
+++ b/test/e2e/script/prepare/setup-e2e/install-yq.sh
@@ -0,0 +1,30 @@
+#!/usr/bin/env bash
+
+# ----------------------------------------------------------------------------
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+# ----------------------------------------------------------------------------
+
+BASE_DIR=$1
+BIN_DIR=$2
+
+if ! command -v yq &> /dev/null; then
+ mkdir -p $BASE_DIR/yq && cd $BASE_DIR/yq
+ curl -kLo yq.tar.gz https://github.com/mikefarah/yq/archive/v4.33.3.tar.gz
+ tar -zxf yq.tar.gz --strip=1
+ go install && go build -ldflags -s && cp yq $BIN_DIR/
+fi
\ No newline at end of file
diff --git a/test/e2e/script/prepare/setup-e2e/install.sh b/test/e2e/script/prepare/setup-e2e/install.sh
new file mode 100755
index 0000000..dade3db
--- /dev/null
+++ b/test/e2e/script/prepare/setup-e2e/install.sh
@@ -0,0 +1,35 @@
+#!/usr/bin/env bash
+
+# ----------------------------------------------------------------------------
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+# ----------------------------------------------------------------------------
+
+set -ex
+
+NAME=$1
+CURRENT_DIR="$(cd "$(dirname $0)"; pwd)"
+
+# prepare base dir
+TMP_DIR=/tmp/skywalking-infra-e2e
+BIN_DIR=$TMP_DIR/bin
+mkdir -p $TMP_DIR $BIN_DIR && cd $TMP_DIR
+
+# execute install
+bash $CURRENT_DIR/install-$NAME.sh $TMP_DIR $BIN_DIR
+
+echo "success to install $NAME"
\ No newline at end of file
diff --git a/test/e2e/script/prepare/setup-oap/log4j2.xml b/test/e2e/script/prepare/setup-oap/log4j2.xml
new file mode 100644
index 0000000..2d1b4e4
--- /dev/null
+++ b/test/e2e/script/prepare/setup-oap/log4j2.xml
@@ -0,0 +1,33 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+ ~ Licensed to the Apache Software Foundation (ASF) under one or more
+ ~ contributor license agreements. See the NOTICE file distributed with
+ ~ this work for additional information regarding copyright ownership.
+ ~ The ASF licenses this file to You under the Apache License, Version 2.0
+ ~ (the "License"); you may not use this file except in compliance with
+ ~ the License. You may obtain a copy of the License at
+ ~
+ ~ http://www.apache.org/licenses/LICENSE-2.0
+ ~
+ ~ Unless required by applicable law or agreed to in writing, software
+ ~ distributed under the License is distributed on an "AS IS" BASIS,
+ ~ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ ~ See the License for the specific language governing permissions and
+ ~ limitations under the License.
+ ~
+ -->
+
+<Configuration status="INFO">
+ <Appenders>
+ <Console name="Console" target="SYSTEM_OUT">
+ <PatternLayout charset="UTF-8" pattern="%d %c %L [%t] %-5p %x - %m%n"/>
+ </Console>
+ </Appenders>
+ <Loggers>
+ <logger name="org.apache.skywalking.oap.server.library.client.jdbc.hikaricp.JDBCClient" level="DEBUG" />
+ <logger name="org.apache.skywalking.oap.server.storage.plugin.jdbc" level="DEBUG" />
+ <Root level="INFO">
+ <AppenderRef ref="Console"/>
+ </Root>
+ </Loggers>
+</Configuration>
diff --git a/test/plugins/runner-helper/templates/windows-docker-compose.tpl b/test/plugins/runner-helper/templates/windows-docker-compose.tpl
index cc85407..befd6f6 100644
--- a/test/plugins/runner-helper/templates/windows-docker-compose.tpl
+++ b/test/plugins/runner-helper/templates/windows-docker-compose.tpl
@@ -36,7 +36,7 @@ services:
condition: service_healthy
volumes:
- {{.Context.WorkSpaceDir}}:/workspace
- command: ["/bin/bash", "-c", "sleep 5s && /workspace/validator.sh"]
+ command: ["/bin/bash", "-c", "sleep 10s && /workspace/validator.sh"]
{{- range $name, $service := .Context.Config.Dependencies }}
{{$name}}:
image: {{$service.Image}}
diff --git a/test/plugins/runner-helper/templates/windows-scenarios.tpl b/test/plugins/runner-helper/templates/windows-scenarios.tpl
index 049e60e..a855e25 100644
--- a/test/plugins/runner-helper/templates/windows-scenarios.tpl
+++ b/test/plugins/runner-helper/templates/windows-scenarios.tpl
@@ -30,6 +30,8 @@ echo "Building the project..."
eval $build_shell
export SW_AGENT_NAME=${project_name}
export SW_AGENT_REPORTER_GRPC_BACKEND_SERVICE=localhost:19876
+export SW_AGENT_METER_COLLECT_INTERVAL=1
+export SW_AGENT_REPORTER_CHECK_INTERVAL=5
eval "$(grep '^export ' ./bin/startup.sh)"
./${project_name} &
diff --git a/test/plugins/scenarios/goframe/plugin.yml b/test/plugins/scenarios/goframe/plugin.yml
index 3d07b78..91f30e9 100644
--- a/test/plugins/scenarios/goframe/plugin.yml
+++ b/test/plugins/scenarios/goframe/plugin.yml
@@ -20,7 +20,7 @@ start-script: ./bin/startup.sh
framework: github.com/gogf/gf
export-port: 8080
support-version:
- - go: 1.18
+ - go: 1.19
framework:
- v2.7.3
- v2.7.2
diff --git a/test/plugins/scenarios/gorm/plugin.yml b/test/plugins/scenarios/gorm/plugin.yml
index 0e01494..826f7cd 100644
--- a/test/plugins/scenarios/gorm/plugin.yml
+++ b/test/plugins/scenarios/gorm/plugin.yml
@@ -20,7 +20,7 @@ start-script: ./bin/startup.sh
framework: gorm.io/gorm
export-port: 8080
support-version:
- - go: 1.17
+ - go: 1.19
framework:
- v1.22.0
- v1.23.0
diff --git a/tools/go-agent/config/agent.default.yaml b/tools/go-agent/config/agent.default.yaml
index 8e14390..f07f22d 100644
--- a/tools/go-agent/config/agent.default.yaml
+++ b/tools/go-agent/config/agent.default.yaml
@@ -34,20 +34,22 @@ agent:
ignore_suffix: ${SW_AGENT_IGNORE_SUFFIX:.jpg,.jpeg,.js,.css,.png,.bmp,.gif,.ico,.mp3,.mp4,.html,.svg}
# If the operation name of the first span is matching, this segment should be ignored.(multiple split by ",").
# Matching rules follow Ant Path match style, like /path/*, /path/**, /path/?.
- # "/path/*" means matching any path that starts with "/path/".
+ # "/path/*" means matching any path that starts with "/path/".
# "/path/**" means matching any path that starts with "/path/" and includes its subpaths.
# "/path/?" means matching any path that starts with "/path/" and has any single character as a wildcard.
trace_ignore_path: ${SW_AGENT_TRACE_IGNORE_PATH:}
reporter:
discard: ${SW_AGENT_REPORTER_DISCARD:false}
+ # Reporter type: "grpc" or "kafka"
+ type: ${SW_AGENT_REPORTER_TYPE:grpc}
+ # The interval(s) of checking service and backend service
+ check_interval: ${SW_AGENT_REPORTER_CHECK_INTERVAL:20}
grpc:
# The gRPC server address of the backend service.
backend_service: ${SW_AGENT_REPORTER_GRPC_BACKEND_SERVICE:127.0.0.1:11800}
# The maximum count of segment for reporting tracing data.
max_send_queue: ${SW_AGENT_REPORTER_GRPC_MAX_SEND_QUEUE:5000}
- # The interval(s) of checking service and backend service
- check_interval: ${SW_AGENT_REPORTER_GRPC_CHECK_INTERVAL:20}
# The authentication string for communicate with backend.
authentication: ${SW_AGENT_REPORTER_GRPC_AUTHENTICATION:}
# The interval(s) of fetching dynamic configuration from backend.
@@ -63,6 +65,27 @@ reporter:
client_cert_chain_path: ${SW_AGENT_REPORTER_GRPC_TLS_CLIENT_CERT_CHAIN_PATH:}
# Controls whether a client verifies the server's certificate chain and host name.
insecure_skip_verify: ${SW_AGENT_REPORTER_GRPC_TLS_INSECURE_SKIP_VERIFY:false}
+ kafka:
+ # Kafka broker addresses, comma separated
+ brokers: ${SW_AGENT_REPORTER_KAFKA_BROKERS:127.0.0.1:9092}
+ # Topic for segments data
+ topic_segment: ${SW_AGENT_REPORTER_KAFKA_TOPIC_SEGMENT:skywalking-segments}
+ # Topic for meters data
+ topic_meter: ${SW_AGENT_REPORTER_KAFKA_TOPIC_METER:skywalking-meters}
+ # Topic for logging data
+ topic_logging: ${SW_AGENT_REPORTER_KAFKA_TOPIC_LOGGING:skywalking-logs}
+ # Topic for management data
+ topic_management: ${SW_AGENT_REPORTER_KAFKA_TOPIC_MANAGEMENT:skywalking-managements}
+ # Send queue size
+ max_send_queue: ${SW_AGENT_REPORTER_KAFKA_MAX_SEND_QUEUE:5000}
+ # Batch size
+ batch_size: ${SW_AGENT_REPORTER_KAFKA_BATCH_SIZE:1000}
+ # Batch bytes
+ batch_bytes: ${SW_AGENT_REPORTER_KAFKA_BATCH_BYTES:1048576}
+ # Batch timeout millis
+ batch_timeout_millis: ${SW_AGENT_REPORTER_KAFKA_BATCH_TIMEOUT_MILLIS:1000}
+ # Acknowledge, 0: none, 1: leader, -1: all
+ acks: ${SW_AGENT_REPORTER_KAFKA_ACKS:1}
log:
# The type determines which logging type is currently used by the system.
@@ -106,7 +129,7 @@ plugin:
redis:
# Limit the bytes size of redis args request
max_args_bytes: ${SW_AGENT_PLUGIN_CONFIG_REDIS_MAX_ARGS_BYTES:1024}
- gin:
+ gin:
# Collect the http header of gin request
collect_request_headers: ${SW_AGENT_PLUGIN_CONFIG_GIN_COLLECT_REQUEST_HEADERS:}
# Controlling the length limitation of all header values
diff --git a/tools/go-agent/config/loader.go b/tools/go-agent/config/loader.go
index 466626a..adc2a11 100644
--- a/tools/go-agent/config/loader.go
+++ b/tools/go-agent/config/loader.go
@@ -55,8 +55,11 @@ type Agent struct {
}
type Reporter struct {
- Discard StringValue `yaml:"discard"`
- GRPC GRPCReporter `yaml:"grpc"`
+ Discard StringValue `yaml:"discard"`
+ Type StringValue `yaml:"type"`
+ CheckInterval StringValue `yaml:"check_interval"`
+ GRPC GRPCReporter `yaml:"grpc"`
+ Kafka KafkaReporter `yaml:"kafka"`
}
type Log struct {
@@ -82,7 +85,6 @@ type Meter struct {
type GRPCReporter struct {
BackendService StringValue `yaml:"backend_service"`
MaxSendQueue StringValue `yaml:"max_send_queue"`
- CheckInterval StringValue `yaml:"check_interval"`
Authentication StringValue `yaml:"authentication"`
CDSFetchInterval StringValue `yaml:"cds_fetch_interval"`
TLS GRPCReporterTLS `yaml:"tls"`
@@ -96,6 +98,19 @@ type GRPCReporterTLS struct {
InsecureSkipVerify StringValue `yaml:"insecure_skip_verify"`
}
+type KafkaReporter struct {
+ Brokers StringValue `yaml:"brokers"`
+ TopicSegment StringValue `yaml:"topic_segment"`
+ TopicMeter StringValue `yaml:"topic_meter"`
+ TopicLogging StringValue `yaml:"topic_logging"`
+ TopicManagement StringValue `yaml:"topic_management"`
+ MaxSendQueue StringValue `yaml:"max_send_queue"`
+ BatchSize StringValue `yaml:"batch_size"`
+ BatchBytes StringValue `yaml:"batch_bytes"`
+ BatchTimeoutMillis StringValue `yaml:"batch_timeout_millis"`
+ Acks StringValue `yaml:"acks"`
+}
+
type Plugin struct {
Config PluginConfig `yaml:"config"`
Excluded StringValue `yaml:"excluded"`
@@ -278,13 +293,18 @@ func (s *StringValue) ToGoBoolFunction() string {
return fmt.Sprintf("func() bool { return %s }", s.ToGoBoolValue())
}
-func (s *StringValue) GetListStringResult() []string {
+func (s *StringValue) GetStringResult() string {
val := s.Default
if s.EnvKey != "" {
if envValue := os.Getenv(s.EnvKey); envValue != "" {
val = envValue
}
}
+ return val
+}
+
+func (s *StringValue) GetListStringResult() []string {
+ val := s.GetStringResult()
if val == "" {
return nil
}
diff --git a/tools/go-agent/instrument/agentcore/instrument.go b/tools/go-agent/instrument/agentcore/instrument.go
index b3e633f..b4b717f 100644
--- a/tools/go-agent/instrument/agentcore/instrument.go
+++ b/tools/go-agent/instrument/agentcore/instrument.go
@@ -168,7 +168,7 @@ import (
var {{.GetGlobalLoggerLinkMethod}} func() interface{}
func (t *Tracer) InitTracer(extend map[string]interface{}) {
- rep, err := reporter.{{.GRPCReporterFuncName}}(t.Log)
+ rep, err := reporter.{{.ReporterInitFuncName}}(t.Log)
if err != nil {
t.Log.Errorf("cannot initialize the reporter: %v", err)
return
@@ -192,11 +192,11 @@ func (t *Tracer) InitTracer(extend map[string]interface{}) {
t.Log.Errorf("cannot initialize the SkyWalking Tracer: %v", err)
}
}`, struct {
- GRPCReporterFuncName string
+ ReporterInitFuncName string
GetGlobalLoggerLinkMethod string
Config *config.Config
}{
- GRPCReporterFuncName: consts.GRPCInitFuncName,
+ ReporterInitFuncName: consts.ReporterInitFuncName,
GetGlobalLoggerLinkMethod: consts.GlobalLoggerGetMethodName,
Config: config.GetConfig(),
})))
diff --git a/tools/go-agent/instrument/consts/reporter.go b/tools/go-agent/instrument/consts/reporter.go
index b090549..6302ec1 100644
--- a/tools/go-agent/instrument/consts/reporter.go
+++ b/tools/go-agent/instrument/consts/reporter.go
@@ -18,5 +18,7 @@
package consts
const (
- GRPCInitFuncName = "GRPCReporterInit"
+ ReporterInitFuncName = "ReporterInit"
+ GrpcReporter = "grpc"
+ KafkaReporter = "kafka"
)
diff --git a/tools/go-agent/instrument/instrument.go b/tools/go-agent/instrument/instrument.go
index a0fe334..4b8cec6 100644
--- a/tools/go-agent/instrument/instrument.go
+++ b/tools/go-agent/instrument/instrument.go
@@ -42,7 +42,7 @@ import (
var instruments = []api.Instrument{
runtime.NewInstrument(),
agentcore.NewInstrument(),
- reporter.NewGRPCInstrument(),
+ reporter.NewInstrument(),
entry.NewInstrument(),
logger.NewInstrument(),
plugins.NewInstrument(),
diff --git a/tools/go-agent/instrument/reporter/instrument.go b/tools/go-agent/instrument/reporter/instrument.go
index 989299f..266a994 100644
--- a/tools/go-agent/instrument/reporter/instrument.go
+++ b/tools/go-agent/instrument/reporter/instrument.go
@@ -34,33 +34,42 @@ import (
"github.com/dave/dst/dstutil"
)
-type GRPCInstrument struct {
+type Instrument struct {
hasToEnhance bool
compileOpts *api.CompileOptions
}
-func NewGRPCInstrument() *GRPCInstrument {
- return &GRPCInstrument{}
+func NewInstrument() *Instrument {
+ return &Instrument{}
}
-func (i *GRPCInstrument) CouldHandle(opts *api.CompileOptions) bool {
+func (i *Instrument) CouldHandle(opts *api.CompileOptions) bool {
i.compileOpts = opts
return opts.Package == "github.com/apache/skywalking-go/agent/reporter"
}
-func (i *GRPCInstrument) FilterAndEdit(path string, curFile *dst.File, cursor *dstutil.Cursor, allFiles []*dst.File) bool {
+func (i *Instrument) FilterAndEdit(path string, curFile *dst.File, cursor *dstutil.Cursor, allFiles []*dst.File) bool {
if i.hasToEnhance {
return false
}
- i.hasToEnhance = true
+ fileName := filepath.Base(path)
+ reporterType := i.getReporterTypeConfig()
+ if fileName == "imports.go" && reporterType == consts.GrpcReporter {
+ tools.DeletePackageImports(curFile,
+ "github.com/segmentio/kafka-go",
+ "github.com/segmentio/kafka-go/compress",
+ "google.golang.org/protobuf/proto")
+ i.hasToEnhance = true
+ }
return true
}
-func (i *GRPCInstrument) AfterEnhanceFile(fromPath, newPath string) error {
+func (i *Instrument) AfterEnhanceFile(fromPath, newPath string) error {
return nil
}
-func (i *GRPCInstrument) WriteExtraFiles(dir string) ([]string, error) {
+func (i *Instrument) WriteExtraFiles(dir string) ([]string, error) {
+ reporterType := i.getReporterTypeConfig()
// copy reporter api files
results := make([]string, 0)
copiedFiles, err := tools.CopyGoFiles(core.FS, "reporter", dir, func(entry fs.DirEntry, f *dst.File) (*tools.DebugInfo, error) {
@@ -70,16 +79,49 @@ func (i *GRPCInstrument) WriteExtraFiles(dir string) ([]string, error) {
debugPath := filepath.Join(i.compileOpts.DebugDir, "plugins", "core", "reporter", entry.Name())
return tools.BuildDSTDebugInfo(debugPath, nil)
}, func(file *dst.File) {
+ pkgUpdates := make(map[string]string)
+ for _, p := range agentcore.CopiedSubPackages {
+ key := strings.ReplaceAll(filepath.Join(agentcore.EnhanceFromBasePackage, p), `\`, `/`)
+ val := strings.ReplaceAll(filepath.Join(agentcore.EnhanceBasePackage, p), `\`, `/`)
+ pkgUpdates[key] = val
+ }
+ tools.ChangePackageImportPath(file, pkgUpdates)
})
if err != nil {
return nil, err
}
results = append(results, copiedFiles...)
- // copy reporter implementations
- // Force the use of '/' delimiter on all platforms
- reporterDirName := strings.ReplaceAll(filepath.Join("reporter", "grpc"), `\`, `/`)
- copiedFiles, err = tools.CopyGoFiles(core.FS, reporterDirName, dir, func(entry fs.DirEntry, f *dst.File) (*tools.DebugInfo, error) {
+ copiedFiles, err = i.copyReporterFiles(dir, reporterType)
+ if err != nil {
+ return nil, err
+ }
+ results = append(results, copiedFiles...)
+
+ // generate the file for export the reporter
+ file, err := i.generateReporterInitFile(dir, reporterType)
+ if err != nil {
+ return nil, err
+ }
+ results = append(results, file)
+
+ return results, nil
+}
+
+func (i *Instrument) getReporterTypeConfig() string {
+ reporterType := config.GetConfig().Reporter.Type.GetStringResult()
+ if reporterType == consts.KafkaReporter {
+ return consts.KafkaReporter
+ }
+ return consts.GrpcReporter
+}
+
+// copy reporter implementations
+// Force the use of '/' delimiter on all platforms
+func (i *Instrument) copyReporterFiles(targetDir, reporterType string) ([]string, error) {
+ copiedFilesResult := make([]string, 0)
+ reporterDirName := strings.ReplaceAll(filepath.Join("reporter", reporterType), `\`, `/`)
+ copiedFiles, err := tools.CopyGoFiles(core.FS, reporterDirName, targetDir, func(entry fs.DirEntry, f *dst.File) (*tools.DebugInfo, error) {
if i.compileOpts.DebugDir == "" {
return nil, nil
}
@@ -96,23 +138,46 @@ func (i *GRPCInstrument) WriteExtraFiles(dir string) ([]string, error) {
tools.ChangePackageImportPath(file, pkgUpdates)
tools.DeletePackageImports(file, "github.com/apache/skywalking-go/plugins/core/reporter")
})
+
if err != nil {
return nil, err
}
- results = append(results, copiedFiles...)
+ copiedFilesResult = append(copiedFilesResult, copiedFiles...)
+ return copiedFilesResult, nil
+}
- // generate the file for export the reporter
- file, err := i.generateReporterInitFile(dir)
+func (i *Instrument) generateReporterInitFile(dir, reporterType string) (string, error) {
+ reporterInitTemplate := baseReporterInitTemplate
+ if reporterType == consts.KafkaReporter {
+ reporterInitTemplate += `
+ _, cdsManager, err := initManager(logger, checkInterval)
if err != nil {
return nil, err
}
- results = append(results, file)
-
- return results, nil
+ return initKafkaReporter(logger, checkInterval, cdsManager)
+}`
+ reporterInitTemplate += kafkaReporterInitFunc
+ } else {
+ reporterInitTemplate += `
+ connManager, cdsManager, err := initManager(logger, checkInterval)
+ if err != nil {
+ return nil, err
+ }
+ return initGRPCReporter(logger, checkInterval, connManager, cdsManager)
+}`
+ reporterInitTemplate += grpcReporterInitFunc
+ }
+ reporterInitTemplate += initManagerFunc
+ return tools.WriteFile(dir, "reporter_init.go", html.UnescapeString(tools.ExecuteTemplate(reporterInitTemplate, struct {
+ InitFuncName string
+ Config *config.Config
+ }{
+ InitFuncName: consts.ReporterInitFuncName,
+ Config: config.GetConfig(),
+ })))
}
-func (i *GRPCInstrument) generateReporterInitFile(dir string) (string, error) {
- return tools.WriteFile(dir, "grpc_init.go", html.UnescapeString(tools.ExecuteTemplate(`package reporter
+const baseReporterInitTemplate = `package reporter
import (
"github.com/apache/skywalking-go/agent/core/operator"
@@ -127,14 +192,20 @@ func {{.InitFuncName}}(logger operator.LogOperator) (Reporter, error) {
if {{.Config.Reporter.Discard.ToGoBoolValue}} {
return NewDiscardReporter(), nil
}
- var opts []ReporterOption
- checkIntervalVal := {{.Config.Reporter.GRPC.CheckInterval.ToGoIntValue "the GRPC reporter check interval must be number"}}
- opts = append(opts, WithCheckInterval(time.Second * time.Duration(checkIntervalVal)))
- opts = append(opts, WithMaxSendQueueSize({{.Config.Reporter.GRPC.MaxSendQueue.ToGoIntValue "the GRPC reporter max queue size must be number"}}))
- opts = append(opts, WithAuthentication({{.Config.Reporter.GRPC.Authentication.ToGoStringValue}}))
- cdsFetchIntervalVal := {{.Config.Reporter.GRPC.CDSFetchInterval.ToGoIntValue "the GRPC reporter max queue size must be number"}}
- opts = append(opts, WithCDS(time.Second * time.Duration(cdsFetchIntervalVal)))
+ checkIntervalVal := {{.Config.Reporter.CheckInterval.ToGoIntValue "the reporter check interval must be number"}}
+ checkInterval := time.Second * time.Duration(checkIntervalVal)
+`
+
+const initManagerFunc = `
+func initManager(logger operator.LogOperator, checkInterval time.Duration) (*ConnectionManager, *CDSManager, error) {
+ authenticationVal := {{.Config.Reporter.GRPC.Authentication.ToGoStringValue}}
+ backendServiceVal := {{.Config.Reporter.GRPC.BackendService.ToGoStringValue}}
+
+ var (
+ connManager *ConnectionManager
+ err error
+ )
if {{.Config.Reporter.GRPC.TLS.Enable.ToGoBoolValue}} {
tc, err := generateTLSCredential({{.Config.Reporter.GRPC.TLS.CAPath.ToGoStringValue}},
{{.Config.Reporter.GRPC.TLS.ClientKeyPath.ToGoStringValue}},
@@ -143,16 +214,65 @@ func {{.InitFuncName}}(logger operator.LogOperator) (Reporter, error) {
if err != nil {
panic(fmt.Sprintf("generate go agent tls credential error: %v", err))
}
- opts = append(opts, WithTransportCredentials(tc))
+ connManager, err = NewConnectionManager(logger, checkInterval, backendServiceVal, authenticationVal, tc)
+ } else {
+ connManager, err = NewConnectionManager(logger, checkInterval, backendServiceVal, authenticationVal, nil)
+ }
+ if err != nil {
+ return nil, nil, err
}
- return NewGRPCReporter(logger, {{.Config.Reporter.GRPC.BackendService.ToGoStringValue}}, opts...)
+ cdsFetchIntervalVal := {{.Config.Reporter.GRPC.CDSFetchInterval.ToGoIntValue "the cds fetch interval must be number"}}
+ cdsFetchInterval := time.Second * time.Duration(cdsFetchIntervalVal)
+ cdsManager, err := NewCDSManager(logger, backendServiceVal, cdsFetchInterval, connManager)
+ if err != nil {
+ return nil, nil, err
+ }
+ return connManager, cdsManager, nil
}
-`, struct {
- InitFuncName string
- Config *config.Config
- }{
- InitFuncName: consts.GRPCInitFuncName,
- Config: config.GetConfig(),
- })))
+`
+
+const grpcReporterInitFunc = `
+
+func initGRPCReporter(logger operator.LogOperator,
+ checkInterval time.Duration,
+ connManager *ConnectionManager,
+ cdsManager *CDSManager) (Reporter, error) {
+ var opts []ReporterOption
+ maxSendQueueVal := {{.Config.Reporter.GRPC.MaxSendQueue.ToGoIntValue "the GRPC reporter max queue size must be number"}}
+ opts = append(opts, WithMaxSendQueueSize(maxSendQueueVal))
+
+ backendServiceVal := {{.Config.Reporter.GRPC.BackendService.ToGoStringValue}}
+ return NewGRPCReporter(logger, backendServiceVal, checkInterval, connManager, cdsManager, opts...)
+}
+`
+
+const kafkaReporterInitFunc = `
+
+func initKafkaReporter(logger operator.LogOperator, checkInterval time.Duration, cdsManager *CDSManager) (Reporter, error) {
+ var opts []ReporterOptionKafka
+
+ topicSegment := {{.Config.Reporter.Kafka.TopicSegment.ToGoStringValue}}
+ opts = append(opts, WithKafkaTopicSegment(topicSegment))
+ topicMeter := {{.Config.Reporter.Kafka.TopicMeter.ToGoStringValue}}
+ opts = append(opts, WithKafkaTopicMeter(topicMeter))
+ topicLogging := {{.Config.Reporter.Kafka.TopicLogging.ToGoStringValue}}
+ opts = append(opts, WithKafkaTopicLogging(topicLogging))
+ topicManagement := {{.Config.Reporter.Kafka.TopicManagement.ToGoStringValue}}
+ opts = append(opts, WithKafkaTopicManagement(topicManagement))
+
+ maxSendQueueVal := {{.Config.Reporter.Kafka.MaxSendQueue.ToGoIntValue "the Kafka reporter max queue size must be a number"}}
+ opts = append(opts, WithKafkaMaxSendQueueSize(maxSendQueueVal))
+ batchSizeVal := {{.Config.Reporter.Kafka.BatchSize.ToGoIntValue "the Kafka reporter batch size must be a number"}}
+ opts = append(opts, WithKafkaBatchSize(batchSizeVal))
+ batchBytesVal := {{.Config.Reporter.Kafka.BatchBytes.ToGoIntValue "the Kafka reporter batch bytes must be a number"}}
+ opts = append(opts, WithKafkaBatchBytes(int64(batchBytesVal)))
+ batchTimeoutMillisVal := {{.Config.Reporter.Kafka.BatchTimeoutMillis.ToGoIntValue "the Kafka reporter batch timeout must be a number"}}
+ opts = append(opts, WithKafkaBatchTimeoutMillis(batchTimeoutMillisVal))
+ acksVal := {{.Config.Reporter.Kafka.Acks.ToGoIntValue "the Kafka reporter acks must be a number"}}
+ opts = append(opts, WithKafkaAcks(acksVal))
+
+ brokers := {{.Config.Reporter.Kafka.Brokers.ToGoStringValue}}
+ return NewKafkaReporter(logger, brokers, checkInterval, cdsManager, opts...)
}
+`