kazusa
/
skywalking
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Support export `Trace` and `Log` through Kafka. (#9817)

This commit is contained in:
Wan Kai 2022-10-21 08:04:18 +08:00 committed by GitHub
parent 8d49b88259
commit d98775790a
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
35 changed files with 1002 additions and 155 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
.github/workflows/skywalking.yaml vendored
View File

@ -584,6 +584,9 @@ jobs:
config: test/e2e-v2/cases/zipkin/mysql/sharding/e2e.yaml
- name: APISIX metrics
config: test/e2e-v2/cases/apisix/otel-collector/e2e.yaml
- name: Exporter Kafka
config: test/e2e-v2/cases/exporter/kafka/e2e.yaml
steps:
- uses: actions/checkout@v3
with:

1
docs/en/changes/changes.md
View File

@ -63,6 +63,7 @@
* Optimize the query time of tasks in ProfileTaskCache.
* Fix metrics was put into wrong slot of the window in the alerting kernel.
* Support `sumPerMinLabeled` in `MAL`.
* Support export `Trace` and `Log` through Kafka.
#### UI

11
docs/en/setup/backend/configuration-vocabulary.md
View File

@ -292,8 +292,15 @@ The Configuration Vocabulary lists all available configurations provided by `app
| - | - | password | Nacos Auth password. | SW_CONFIG_NACOS_PASSWORD | - |
| - | - | accessKey | Nacos Auth accessKey. | SW_CONFIG_NACOS_ACCESSKEY | - |
| - | - | secretKey | Nacos Auth secretKey. | SW_CONFIG_NACOS_SECRETKEY | - |
| exporter | grpc | targetHost | The host of target gRPC server for receiving export data. | SW_EXPORTER_GRPC_HOST | 127.0.0.1 |
| - | - | targetPort | The port of target gRPC server for receiving export data. | SW_EXPORTER_GRPC_PORT | 9870 |
| exporter | default | enableGRPCMetrics | Enable gRPC metrics exporter. | SW_EXPORTER_ENABLE_GRPC_METRICS | false |
| - | - | gRPCTargetHost | The host of target gRPC server for receiving export data | SW_EXPORTER_GRPC_HOST | 127.0.0.1 |
| - | - | gRPCTargetPort | The port of target gRPC server for receiving export data. | SW_EXPORTER_GRPC_PORT | 9870 |
| - | - | enableKafkaTrace | Enable Kafka trace exporter. | SW_EXPORTER_ENABLE_KAFKA_TRACE | false |
| - | - | enableKafkaLog | Enable Kafka log exporter. | SW_EXPORTER_ENABLE_KAFKA_LOG | false |
| - | - | kafkaBootstrapServers | A list of host/port pairs to use for establishing the initial connection to the Kafka cluster. | SW_EXPORTER_KAFKA_SERVERS | localhost:9092 |
| - | - | kafkaProducerConfig | Kafka producer config, JSON format as Properties. | SW_EXPORTER_KAFKA_PRODUCER_CONFIG | - |
| - | - | kafkaTopicTrace | Kafka topic name for trace. | SW_EXPORTER_KAFKA_TOPIC_TRACE | skywalking-export-trace |
| - | - | kafkaTopicLog | Kafka topic name for log. | SW_EXPORTER_KAFKA_TOPIC_LOG | skywalking-export-log |
| health-checker | default | checkIntervalSeconds | The period of checking OAP internal health status (in seconds). | SW_HEALTH_CHECKER_INTERVAL_SECONDS | 5 |
| configuration-discovery | default | disableMessageDigest | If true, agent receives the latest configuration every time, even without making any changes. By default, OAP uses the SHA512 message digest mechanism to detect changes in configuration. | SW_DISABLE_MESSAGE_DIGEST | false |
| receiver-event | default | gRPC services that handle events data. | - | - | |

121
docs/en/setup/backend/exporter.md Normal file
View File

@ -0,0 +1,121 @@
# Exporter
SkyWalking provides the essential functions of observability, including metrics aggregation, trace, log, alerting, and profiling.
In many real-world scenarios, users may want to forward their data to a 3rd party system for further in-depth analysis.
**Exporter** has made that possible.
The exporter is an independent module that has to be manually activated.
Right now, we provide the following exporting channels:
1. gRPC Exporter
- [Metrics](#metrics-grpc-exporter)
1. Kafka Exporter
- [Trace](#trace-kafka-exporter)
- [Log](#log-kafka-exporter)
## gRPC Exporter
### Metrics gRPC Exporter
Metrics gRPC exporter uses SkyWalking's native export service definition. Here is the proto definition: [metric-exporter.proto](https://github.com/apache/skywalking/blob/master/oap-server/exporter/src/main/proto/metric-exporter.proto).
```proto
service MetricExportService {
rpc export (stream ExportMetricValue) returns (ExportResponse) {
}
rpc subscription (SubscriptionReq) returns (SubscriptionsResp) {
}
}
```
To activate the exporter, you should set `${SW_EXPORTER_ENABLE_GRPC_METRICS:true}` and config the target gRPC server address.
```yaml
exporter:
default:
# gRPC exporter
enableGRPCMetrics: ${SW_EXPORTER_ENABLE_GRPC_METRICS:true}
gRPCTargetHost: ${SW_EXPORTER_GRPC_HOST:127.0.0.1}
gRPCTargetPort: ${SW_EXPORTER_GRPC_PORT:9870}
...
```
- `gRPCTargetHost`:`gRPCTargetPort` is the expected target service address. You could set any gRPC server to receive the data.
- Target gRPC service needs to go on standby; otherwise, the OAP startup may fail.
#### Target exporter service
1. Subscription implementation.
Return the expected metrics name list with event type (incremental or total). All names must match the OAL/MAL script definition.
Return empty list, if you want to export all metrics in the incremental event type.
2. Export implementation.
Stream service. All subscribed metrics will be sent here based on the OAP core schedule. Also, if the OAP is deployed as a cluster,
this method will be called concurrently. For metrics value, you need to follow `#type` to choose `#longValue` or `#doubleValue`.
## Kafka Exporter
### Trace Kafka Exporter
Trace kafka exporter pushes messages to the Kafka Broker and Topic `skywalking-trace` to export the trace. Here is the message:
```
ProducerRecord<String, Bytes>
Key: TraceSegmentId
Value: Bytes of SegmentObject
```
The `SegmentObject` definition follows the protocol:
[SkyWalking data collect protocol#Tracing.proto](https://github.com/apache/skywalking-data-collect-protocol/blob/master/language-agent/Tracing.proto).
```proto
// The segment is a collection of spans. It includes all collected spans in a simple one request context, such as a HTTP request process.
message SegmentObject {
string traceId = 1;
string traceSegmentId = 2;
repeated SpanObject spans = 3;
string service = 4;
string serviceInstance = 5;
bool isSizeLimited = 6;
}
```
To activate the exporter, you should set `${SW_EXPORTER_ENABLE_KAFKA_TRACE:true}` and config the Kafka server.
```yaml
exporter:
default:
# Kafka exporter
enableKafkaTrace: ${SW_EXPORTER_ENABLE_KAFKA_TRACE:true}
kafkaBootstrapServers: ${SW_EXPORTER_KAFKA_SERVERS:localhost:9092}
# Kafka producer config, JSON format as Properties.
kafkaProducerConfig: ${SW_EXPORTER_KAFKA_PRODUCER_CONFIG:""}
kafkaTopicTrace: ${SW_EXPORTER_KAFKA_TOPIC_TRACE:skywalking-trace}
...
```
### Log Kafka Exporter
Log kafka exporter pushes messages to the Kafka Broker and Topic `skywalking-log` to export the log. Here is the message:
```
ProducerRecord<String, Bytes>
Key: LogRecordId
Value: Bytes of LogData
```
The `LogData` definition follows the protocol:
[SkyWalking data collect protocol#Logging.proto](https://github.com/apache/skywalking-data-collect-protocol/blob/master/logging/Logging.proto).
```proto
message LogData {
int64 timestamp = 1;
string service = 2;
string serviceInstance = 3;
string endpoint = 4;
LogDataBody body = 5;
TraceContext traceContext = 6;
LogTags tags = 7;
string layer = 8;
}
```
To activate the exporter, you should set `${SW_EXPORTER_ENABLE_KAFKA_LOG:true}` and config the Kafka server.
```yaml
exporter:
default:
# Kafka exporter
enableKafkaLog: ${SW_EXPORTER_ENABLE_KAFKA_LOG:true}
kafkaBootstrapServers: ${SW_EXPORTER_KAFKA_SERVERS:localhost:9092}
# Kafka producer config, JSON format as Properties.
kafkaProducerConfig: ${SW_EXPORTER_KAFKA_PRODUCER_CONFIG:""}
kafkaTopicLog: ${SW_EXPORTER_KAFKA_TOPIC_LOG:skywalking-log}
...
```

82
docs/en/setup/backend/metrics-exporter.md
View File

@ -1,81 +1 @@
# Metrics Exporter
SkyWalking provides the essential functions of metrics aggregation, alarm, and analysis.
In many real-world scenarios, users may want to forward their data to a 3rd party system for further in-depth analysis.
**Metrics Exporter** has made that possible.
The metrics exporter is an independent module that has to be manually activated.
Right now, we provide the following exporters:
1. gRPC exporter
## gRPC exporter
gRPC exporter uses SkyWalking's native exporter service definition. Here is the proto definition.
```proto
service MetricExportService {
rpc export (stream ExportMetricValue) returns (ExportResponse) {
}
rpc subscription (SubscriptionReq) returns (SubscriptionsResp) {
}
}
message ExportMetricValue {
string metricName = 1;
string entityName = 2;
string entityId = 3;
ValueType type = 4;
int64 timeBucket = 5;
int64 longValue = 6;
double doubleValue = 7;
repeated int64 longValues = 8;
}
message SubscriptionsResp {
repeated SubscriptionMetric metrics = 1;
}
message SubscriptionMetric {
string metricName = 1;
EventType eventType = 2;
}
enum ValueType {
LONG = 0;
DOUBLE = 1;
MULTI_LONG = 2;
}
enum EventType {
// The metrics aggregated in this bulk, not include the existing persistent data.
INCREMENT = 0;
// Final result of the metrics at this moment.
TOTAL = 1;
}
message SubscriptionReq {
}
message ExportResponse {
}
```
To activate the exporter, you should add this into your `application.yml`
```yaml
exporter:
grpc:
targetHost: 127.0.0.1
targetPort: 9870
```
- `targetHost`:`targetPort` is the expected target service address. You could set any gRPC server to receive the data.
- Target gRPC service needs to go on standby; otherwise, the OAP startup may fail.
## Target exporter service
### Subscription implementation
Return the expected metrics name list with event type (incremental or total). All names must match the OAL/MAL script definition.
Return empty list, if you want to export all metrics in the incremental event type.
### Export implementation
Stream service. All subscribed metrics will be sent here based on the OAP core schedule. Also, if the OAP is deployed as a cluster,
this method will be called concurrently. For metrics value, you need to follow `#type` to choose `#longValue` or `#doubleValue`.
All SkyWalking exporter(metrics, trace, log) instructions had been moved [here](exporter.md).

4
docs/menu.yml
View File

@ -133,8 +133,8 @@ catalog:
path: "/en/setup/backend/on-demand-pod-log"
- name: "Extension"
catalog:
- name: "Metrics Exporter"
path: "/en/setup/backend/metrics-exporter"
- name: "Exporter"
path: "/en/setup/backend/exporter"
- name: "Dynamic Configuration"
path: "/en/setup/backend/dynamic-config"
- name: "UI Setup"

14
oap-server-bom/pom.xml
View File

@ -73,6 +73,8 @@
<httpcore.version>4.4.13</httpcore.version>
<commons-compress.version>1.21</commons-compress.version>
<banyandb-java-client.version>0.1.0</banyandb-java-client.version>
<kafka-clients.version>2.4.1</kafka-clients.version>
<spring-kafka-test.version>2.4.6.RELEASE</spring-kafka-test.version>
</properties>
<dependencyManagement>
@ -554,6 +556,18 @@
<artifactId>commons-compress</artifactId>
<version>${commons-compress.version}</version>
</dependency>
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka-clients</artifactId>
<version>${kafka-clients.version}</version>
</dependency>
<dependency>
<groupId>org.springframework.kafka</groupId>
<artifactId>spring-kafka-test</artifactId>
<version>${spring-kafka-test.version}</version>
<scope>test</scope>
</dependency>
</dependencies>
</dependencyManagement>
</project>

6
oap-server/exporter/pom.xml
View File

@ -38,6 +38,10 @@
<artifactId>library-datacarrier-queue</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka-clients</artifactId>
</dependency>
<dependency>
<groupId>io.grpc</groupId>
<artifactId>grpc-testing</artifactId>
@ -76,4 +80,4 @@
</plugin>
</plugins>
</build>
</project>
</project>

44
oap-server/exporter/src/main/java/org/apache/skywalking/oap/server/exporter/provider/grpc/GRPCExporterProvider.java → oap-server/exporter/src/main/java/org/apache/skywalking/oap/server/exporter/provider/ExporterProvider.java
View File

@ -16,24 +16,35 @@
*
*/
package org.apache.skywalking.oap.server.exporter.provider.grpc;
package org.apache.skywalking.oap.server.exporter.provider;
import org.apache.skywalking.oap.server.core.CoreModule;
import org.apache.skywalking.oap.server.core.exporter.ExporterModule;
import org.apache.skywalking.oap.server.core.exporter.LogExportService;
import org.apache.skywalking.oap.server.core.exporter.MetricValuesExportService;
import org.apache.skywalking.oap.server.core.exporter.TraceExportService;
import org.apache.skywalking.oap.server.exporter.provider.grpc.GRPCMetricsExporter;
import org.apache.skywalking.oap.server.exporter.provider.kafka.log.KafkaLogExporter;
import org.apache.skywalking.oap.server.exporter.provider.kafka.trace.KafkaTraceExporter;
import org.apache.skywalking.oap.server.library.module.ModuleConfig;
import org.apache.skywalking.oap.server.library.module.ModuleDefine;
import org.apache.skywalking.oap.server.library.module.ModuleProvider;
import org.apache.skywalking.oap.server.library.module.ModuleStartException;
import org.apache.skywalking.oap.server.library.module.ServiceNotProvidedException;
public class GRPCExporterProvider extends ModuleProvider {
private GRPCExporterSetting setting;
private GRPCExporter exporter;
public class ExporterProvider extends ModuleProvider {
private final ExporterSetting setting;
private GRPCMetricsExporter grpcMetricsExporter;
private KafkaTraceExporter kafkaTraceExporter;
private KafkaLogExporter kafkaLogExporter;
public ExporterProvider() {
setting = new ExporterSetting();
}
@Override
public String name() {
return "grpc";
return "default";
}
@Override
@ -43,24 +54,37 @@ public class GRPCExporterProvider extends ModuleProvider {
@Override
public ModuleConfig createConfigBeanIfAbsent() {
setting = new GRPCExporterSetting();
return setting;
}
@Override
public void prepare() throws ServiceNotProvidedException, ModuleStartException {
exporter = new GRPCExporter(setting);
this.registerServiceImplementation(MetricValuesExportService.class, exporter);
grpcMetricsExporter = new GRPCMetricsExporter(setting);
kafkaTraceExporter = new KafkaTraceExporter(getManager(), setting);
kafkaLogExporter = new KafkaLogExporter(getManager(), setting);
this.registerServiceImplementation(MetricValuesExportService.class, grpcMetricsExporter);
this.registerServiceImplementation(TraceExportService.class, kafkaTraceExporter);
this.registerServiceImplementation(LogExportService.class, kafkaLogExporter);
}
@Override
public void start() throws ServiceNotProvidedException, ModuleStartException {
if (setting.isEnableGRPCMetrics()) {
grpcMetricsExporter.start();
}
if (setting.isEnableKafkaTrace()) {
kafkaTraceExporter.start();
}
if (setting.isEnableKafkaLog()) {
kafkaLogExporter.start();
}
}
@Override
public void notifyAfterCompleted() throws ServiceNotProvidedException, ModuleStartException {
exporter.fetchSubscriptionList();
if (setting.isEnableGRPCMetrics()) {
grpcMetricsExporter.fetchSubscriptionList();
}
}
@Override

41
oap-server/exporter/src/main/java/org/apache/skywalking/oap/server/exporter/provider/ExporterSetting.java Normal file
View File

@ -0,0 +1,41 @@
/*
* 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 org.apache.skywalking.oap.server.exporter.provider;
import lombok.Getter;
import lombok.Setter;
import org.apache.skywalking.oap.server.library.module.ModuleConfig;
@Setter
@Getter
public class ExporterSetting extends ModuleConfig {
private boolean enableGRPCMetrics = false;
private String gRPCTargetHost;
private int gRPCTargetPort;
private int bufferChannelSize = 20000;
private int bufferChannelNum = 2;
//kafka
private boolean enableKafkaTrace = false;
private boolean enableKafkaLog = false;
private String kafkaBootstrapServers;
private String kafkaProducerConfig;
private String kafkaTopicTrace = "skywalking-export-trace";
private String kafkaTopicLog = "skywalking-export-log";
}

34
oap-server/exporter/src/main/java/org/apache/skywalking/oap/server/exporter/provider/grpc/GRPCExporter.java → oap-server/exporter/src/main/java/org/apache/skywalking/oap/server/exporter/provider/grpc/GRPCMetricsExporter.java
View File

@ -45,6 +45,7 @@ import org.apache.skywalking.oap.server.exporter.grpc.SubscriptionMetric;
import org.apache.skywalking.oap.server.exporter.grpc.SubscriptionReq;
import org.apache.skywalking.oap.server.exporter.grpc.SubscriptionsResp;
import org.apache.skywalking.oap.server.exporter.grpc.ValueType;
import org.apache.skywalking.oap.server.exporter.provider.ExporterSetting;
import org.apache.skywalking.oap.server.exporter.provider.MetricFormatter;
import org.apache.skywalking.oap.server.library.client.grpc.GRPCClient;
import org.apache.skywalking.oap.server.library.datacarrier.DataCarrier;
@ -52,22 +53,26 @@ import org.apache.skywalking.oap.server.library.datacarrier.consumer.IConsumer;
import org.apache.skywalking.oap.server.library.util.GRPCStreamStatus;
@Slf4j
public class GRPCExporter extends MetricFormatter implements MetricValuesExportService, IConsumer<ExportData> {
public class GRPCMetricsExporter extends MetricFormatter implements MetricValuesExportService, IConsumer<ExportData> {
/**
* The period of subscription list fetching is hardcoded as 30s.
*/
private static final long FETCH_SUBSCRIPTION_PERIOD = 30_000;
private final GRPCExporterSetting setting;
private final MetricExportServiceGrpc.MetricExportServiceStub exportServiceFutureStub;
private final MetricExportServiceGrpc.MetricExportServiceBlockingStub blockingStub;
private final DataCarrier exportBuffer;
private final ReentrantLock fetchListLock;
private final ExporterSetting setting;
private MetricExportServiceGrpc.MetricExportServiceStub exportServiceFutureStub;
private MetricExportServiceGrpc.MetricExportServiceBlockingStub blockingStub;
private DataCarrier exportBuffer;
private ReentrantLock fetchListLock;
private volatile List<SubscriptionMetric> subscriptionList;
private volatile long lastFetchTimestamp = 0;
public GRPCExporter(GRPCExporterSetting setting) {
public GRPCMetricsExporter(ExporterSetting setting) {
this.setting = setting;
GRPCClient client = new GRPCClient(setting.getTargetHost(), setting.getTargetPort());
}
@Override
public void start() {
GRPCClient client = new GRPCClient(setting.getGRPCTargetHost(), setting.getGRPCTargetPort());
client.connect();
ManagedChannel channel = client.getChannel();
exportServiceFutureStub = MetricExportServiceGrpc.newStub(channel);
@ -98,6 +103,11 @@ public class GRPCExporter extends MetricFormatter implements MetricValuesExportS
}
}
@Override
public boolean isEnabled() {
return setting.isEnableGRPCMetrics();
}
/**
* Read the subscription list.
*/
@ -209,17 +219,17 @@ public class GRPCExporter extends MetricFormatter implements MetricValuesExportS
if (sleepTime > 2000L) {
log.warn(
"Export {} metrics to {}:{}, wait {} milliseconds.", exportNum.get(), setting.getTargetHost(),
"Export {} metrics to {}:{}, wait {} milliseconds.", exportNum.get(), setting.getGRPCTargetHost(),
setting
.getTargetPort(), sleepTime
.getGRPCTargetPort(), sleepTime
);
cycle = 2000L;
}
}
log.debug(
"Exported {} metrics to {}:{} in {} milliseconds.", exportNum.get(), setting.getTargetHost(), setting
.getTargetPort(), sleepTime);
"Exported {} metrics to {}:{} in {} milliseconds.", exportNum.get(), setting.getGRPCTargetHost(), setting
.getGRPCTargetPort(), sleepTime);
fetchSubscriptionList();
}

54
oap-server/exporter/src/main/java/org/apache/skywalking/oap/server/exporter/provider/kafka/KafkaExportProducer.java Normal file
View File

@ -0,0 +1,54 @@
/*
* 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 org.apache.skywalking.oap.server.exporter.provider.kafka;
import com.google.gson.Gson;
import java.util.Properties;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.BytesSerializer;
import org.apache.kafka.common.serialization.StringSerializer;
import org.apache.kafka.common.utils.Bytes;
import org.apache.skywalking.oap.server.exporter.provider.ExporterSetting;
import org.apache.skywalking.oap.server.library.util.StringUtil;
@Slf4j
public abstract class KafkaExportProducer {
protected final ExporterSetting setting;
private volatile KafkaProducer<String, Bytes> producer;
public KafkaExportProducer(ExporterSetting setting) {
this.setting = setting;
}
protected KafkaProducer<String, Bytes> getProducer() {
if (producer == null) {
Properties properties = new Properties();
properties.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, setting.getKafkaBootstrapServers());
if (StringUtil.isNotEmpty(setting.getKafkaProducerConfig())) {
Gson gson = new Gson();
Properties override = gson.fromJson(setting.getKafkaProducerConfig(), Properties.class);
properties.putAll(override);
}
producer = new KafkaProducer<>(properties, new StringSerializer(), new BytesSerializer());
}
return producer;
}
}

189
oap-server/exporter/src/main/java/org/apache/skywalking/oap/server/exporter/provider/kafka/log/KafkaLogExporter.java Normal file
View File

@ -0,0 +1,189 @@
/*
* 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 org.apache.skywalking.oap.server.exporter.provider.kafka.log;
import com.google.protobuf.InvalidProtocolBufferException;
import java.util.List;
import java.util.Properties;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.utils.Bytes;
import org.apache.skywalking.apm.network.logging.v3.JSONLog;
import org.apache.skywalking.apm.network.logging.v3.LogData;
import org.apache.skywalking.apm.network.logging.v3.LogDataBody;
import org.apache.skywalking.apm.network.logging.v3.LogTags;
import org.apache.skywalking.apm.network.logging.v3.TextLog;
import org.apache.skywalking.apm.network.logging.v3.TraceContext;
import org.apache.skywalking.apm.network.logging.v3.YAMLLog;
import org.apache.skywalking.oap.server.core.UnexpectedException;
import org.apache.skywalking.oap.server.core.analysis.IDManager;
import org.apache.skywalking.oap.server.core.analysis.manual.log.LogRecord;
import org.apache.skywalking.oap.server.core.exporter.LogExportService;
import org.apache.skywalking.oap.server.core.query.type.ContentType;
import org.apache.skywalking.oap.server.exporter.provider.ExporterSetting;
import org.apache.skywalking.oap.server.exporter.provider.kafka.KafkaExportProducer;
import org.apache.skywalking.oap.server.library.datacarrier.DataCarrier;
import org.apache.skywalking.oap.server.library.datacarrier.buffer.BufferStrategy;
import org.apache.skywalking.oap.server.library.datacarrier.consumer.IConsumer;
import org.apache.skywalking.oap.server.library.module.ModuleManager;
import org.apache.skywalking.oap.server.library.util.StringUtil;
import org.apache.skywalking.oap.server.telemetry.TelemetryModule;
import org.apache.skywalking.oap.server.telemetry.api.CounterMetrics;
import org.apache.skywalking.oap.server.telemetry.api.MetricsCreator;
import org.apache.skywalking.oap.server.telemetry.api.MetricsTag;
@Slf4j
public class KafkaLogExporter extends KafkaExportProducer implements LogExportService, IConsumer<LogRecord> {
private DataCarrier<LogRecord> exportBuffer;
private CounterMetrics successCounter;
private CounterMetrics errorCounter;
private final ModuleManager moduleManager;
public KafkaLogExporter(ModuleManager manager, ExporterSetting setting) {
super(setting);
this.moduleManager = manager;
}
@Override
public void start() {
super.getProducer();
exportBuffer = new DataCarrier<>(
"KafkaLogExporter", "KafkaLogExporter", setting.getBufferChannelNum(), setting.getBufferChannelSize(),
BufferStrategy.IF_POSSIBLE
);
exportBuffer.consume(this, 1, 200);
MetricsCreator metricsCreator = moduleManager.find(TelemetryModule.NAME)
.provider()
.getService(MetricsCreator.class);
successCounter = metricsCreator.createCounter(
"kafka_exporter_log_success_count", "The success number of log exported by kafka exporter.",
new MetricsTag.Keys("protocol"),
new MetricsTag.Values("kafka")
);
errorCounter = metricsCreator.createCounter(
"kafka_exporter_log_error_count", "The error number of log exported by kafka exporter",
new MetricsTag.Keys("protocol"), new MetricsTag.Values("kafka")
);
}
@Override
public void export(final LogRecord logRecord) {
if (logRecord != null) {
exportBuffer.produce(logRecord);
}
}
@Override
public boolean isEnabled() {
return setting.isEnableKafkaLog();
}
@Override
public void init(final Properties properties) {
}
@Override
public void consume(final List<LogRecord> data) {
for (LogRecord logRecord : data) {
if (logRecord != null) {
try {
LogData logData = transLogData(logRecord);
ProducerRecord<String, Bytes> record = new ProducerRecord<>(
setting.getKafkaTopicLog(),
logRecord.id(),
Bytes.wrap(logData.toByteArray())
);
super.getProducer().send(record, (metadata, ex) -> {
if (ex != null) {
errorCounter.inc();
log.error("Failed to export Log.", ex);
} else {
successCounter.inc();
}
});
} catch (InvalidProtocolBufferException e) {
throw new UnexpectedException(
"Failed to parse Log tags from LogRecord, id: " + logRecord.id() + ".", e);
}
}
}
}
@Override
public void onError(final List<LogRecord> data, final Throwable t) {
}
@Override
public void onExit() {
}
private LogData transLogData(LogRecord logRecord) throws InvalidProtocolBufferException {
LogData.Builder builder = LogData.newBuilder();
LogDataBody.Builder bodyBuilder = LogDataBody.newBuilder();
switch (ContentType.instanceOf(logRecord.getContentType())) {
case JSON:
bodyBuilder.setType(ContentType.JSON.name());
bodyBuilder.setJson(JSONLog.newBuilder().setJson(logRecord.getContent()));
break;
case YAML:
bodyBuilder.setType(ContentType.YAML.name());
bodyBuilder.setYaml(YAMLLog.newBuilder().setYaml(logRecord.getContent()));
break;
case TEXT:
bodyBuilder.setType(ContentType.TEXT.name());
bodyBuilder.setText(TextLog.newBuilder().setText(logRecord.getContent()));
break;
case NONE:
bodyBuilder.setType(ContentType.NONE.name());
break;
default:
throw new UnexpectedException(
"Failed to parse Log ContentType value: " + logRecord.getContentType() + " from LogRecord, id: " + logRecord.id() + ".");
}
builder.setBody(bodyBuilder);
builder.setTimestamp(logRecord.getTimestamp());
builder.setService(IDManager.ServiceID.analysisId(logRecord.getServiceId()).getName());
if (StringUtil.isNotEmpty(logRecord.getServiceInstanceId())) {
builder.setServiceInstance(
IDManager.ServiceInstanceID.analysisId(logRecord.getServiceInstanceId()).getName());
}
if (StringUtil.isNotEmpty(logRecord.getEndpointId())) {
builder.setEndpoint(
IDManager.EndpointID.analysisId(logRecord.getEndpointId()).getEndpointName());
}
TraceContext.Builder contextBuilder = TraceContext.newBuilder();
if (StringUtil.isNotEmpty(logRecord.getTraceSegmentId())) {
contextBuilder.setTraceSegmentId(logRecord.getTraceSegmentId());
contextBuilder.setSpanId(logRecord.getSpanId());
}
if (StringUtil.isNotEmpty(logRecord.getTraceId())) {
contextBuilder.setTraceId(logRecord.getTraceId());
}
builder.setTraceContext(contextBuilder);
if (logRecord.getTagsRawData() != null) {
builder.setTags(LogTags.parseFrom(logRecord.getTagsRawData()));
}
return builder.build();
}
}

130
oap-server/exporter/src/main/java/org/apache/skywalking/oap/server/exporter/provider/kafka/trace/KafkaTraceExporter.java Normal file
View File

@ -0,0 +1,130 @@
/*
* 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 org.apache.skywalking.oap.server.exporter.provider.kafka.trace;
import com.google.protobuf.InvalidProtocolBufferException;
import java.util.List;
import java.util.Properties;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.utils.Bytes;
import org.apache.skywalking.apm.network.language.agent.v3.SegmentObject;
import org.apache.skywalking.oap.server.core.UnexpectedException;
import org.apache.skywalking.oap.server.core.analysis.manual.segment.SegmentRecord;
import org.apache.skywalking.oap.server.core.exporter.TraceExportService;
import org.apache.skywalking.oap.server.exporter.provider.ExporterSetting;
import org.apache.skywalking.oap.server.exporter.provider.kafka.KafkaExportProducer;
import org.apache.skywalking.oap.server.library.datacarrier.DataCarrier;
import org.apache.skywalking.oap.server.library.datacarrier.buffer.BufferStrategy;
import org.apache.skywalking.oap.server.library.datacarrier.consumer.IConsumer;
import org.apache.skywalking.oap.server.library.module.ModuleManager;
import org.apache.skywalking.oap.server.telemetry.TelemetryModule;
import org.apache.skywalking.oap.server.telemetry.api.CounterMetrics;
import org.apache.skywalking.oap.server.telemetry.api.MetricsCreator;
import org.apache.skywalking.oap.server.telemetry.api.MetricsTag;
@Slf4j
public class KafkaTraceExporter extends KafkaExportProducer implements TraceExportService, IConsumer<SegmentRecord> {
private DataCarrier<SegmentRecord> exportBuffer;
private CounterMetrics successCounter;
private CounterMetrics errorCounter;
private final ModuleManager moduleManager;
public KafkaTraceExporter(ModuleManager manager, ExporterSetting setting) {
super(setting);
this.moduleManager = manager;
}
@Override
public void start() {
super.getProducer();
exportBuffer = new DataCarrier<>(
"KafkaTraceExporter", "KafkaTraceExporter", setting.getBufferChannelNum(), setting.getBufferChannelSize(),
BufferStrategy.IF_POSSIBLE
);
exportBuffer.consume(this, 1, 200);
MetricsCreator metricsCreator = moduleManager.find(TelemetryModule.NAME)
.provider()
.getService(MetricsCreator.class);
successCounter = metricsCreator.createCounter(
"kafka_exporter_trace_success_count", "The success number of traces exported by kafka exporter.",
new MetricsTag.Keys("protocol"),
new MetricsTag.Values("kafka")
);
errorCounter = metricsCreator.createCounter(
"kafka_exporter_trace_error_count", "The error number of traces exported by kafka exporter",
new MetricsTag.Keys("protocol"), new MetricsTag.Values("kafka")
);
}
public void export(SegmentRecord segmentRecord) {
if (segmentRecord != null) {
exportBuffer.produce(segmentRecord);
}
}
@Override
public boolean isEnabled() {
return setting.isEnableKafkaTrace();
}
@Override
public void init(final Properties properties) {
}
@Override
public void consume(final List<SegmentRecord> data) {
for (SegmentRecord segmentRecord : data) {
if (segmentRecord != null) {
try {
SegmentObject segmentObject = SegmentObject.parseFrom(segmentRecord.getDataBinary());
ProducerRecord<String, Bytes> record = new ProducerRecord<>(
setting.getKafkaTopicTrace(),
segmentObject.getTraceSegmentId(),
Bytes.wrap(segmentObject.toByteArray())
);
super.getProducer().send(record, (metadata, ex) -> {
if (ex != null) {
errorCounter.inc();
log.error("Failed to export Trace.", ex);
} else {
successCounter.inc();
}
});
} catch (InvalidProtocolBufferException e) {
throw new UnexpectedException(
"Failed to parse SegmentObject from SegmentRecord, id: " + segmentRecord.getSegmentId() + ".", e
);
}
}
}
}
@Override
public void onError(final List<SegmentRecord> data, final Throwable t) {
}
@Override
public void onExit() {
}
}

2
oap-server/exporter/src/main/resources/META-INF/services/org.apache.skywalking.oap.server.library.module.ModuleProvider
View File

@ -16,4 +16,4 @@
#
#
org.apache.skywalking.oap.server.exporter.provider.grpc.GRPCExporterProvider
org.apache.skywalking.oap.server.exporter.provider.ExporterProvider

20
oap-server/exporter/src/test/java/org/apache/skywalking/oap/server/exporter/provider/grpc/GRPCExporterProviderTest.java
View File

@ -22,6 +22,8 @@ import java.util.Iterator;
import java.util.ServiceLoader;
import org.apache.skywalking.oap.server.core.CoreModule;
import org.apache.skywalking.oap.server.core.exporter.ExporterModule;
import org.apache.skywalking.oap.server.exporter.provider.ExporterProvider;
import org.apache.skywalking.oap.server.exporter.provider.ExporterSetting;
import org.apache.skywalking.oap.server.library.module.ModuleManager;
import org.apache.skywalking.oap.server.library.module.ModuleProvider;
import org.apache.skywalking.oap.server.library.module.ModuleProviderHolder;
@ -53,17 +55,17 @@ public class GRPCExporterProviderTest {
assertTrue(moduleProviderIterator.hasNext());
grpcExporterProvider = moduleProviderIterator.next();
assertTrue(grpcExporterProvider instanceof GRPCExporterProvider);
assertTrue(grpcExporterProvider instanceof ExporterProvider);
GRPCExporterSetting config = (GRPCExporterSetting) grpcExporterProvider.createConfigBeanIfAbsent();
ExporterSetting config = (ExporterSetting) grpcExporterProvider.createConfigBeanIfAbsent();
assertNotNull(config);
assertNull(config.getTargetHost());
assertEquals(0, config.getTargetPort());
assertNull(config.getGRPCTargetHost());
assertEquals(0, config.getGRPCTargetPort());
assertEquals(20000, config.getBufferChannelSize());
assertEquals(2, config.getBufferChannelNum());
//for test
config.setTargetHost("localhost");
config.setGRPCTargetHost("localhost");
grpcExporterProvider.prepare();
@ -72,7 +74,7 @@ public class GRPCExporterProviderTest {
@Test
public void name() {
assertEquals("grpc", grpcExporterProvider.name());
assertEquals("default", grpcExporterProvider.name());
}
@Test
@ -82,7 +84,7 @@ public class GRPCExporterProviderTest {
@Test
public void notifyAfterCompleted() throws ServiceNotProvidedException, ModuleStartException {
GRPCExporter exporter = mock(GRPCExporter.class);
GRPCMetricsExporter exporter = mock(GRPCMetricsExporter.class);
ModuleManager manager = mock(ModuleManager.class);
ModuleProviderHolder providerHolder = mock(ModuleProviderHolder.class);
@ -95,7 +97,7 @@ public class GRPCExporterProviderTest {
doNothing().when(exporter).fetchSubscriptionList();
grpcExporterProvider.setManager(manager);
Whitebox.setInternalState(grpcExporterProvider, "exporter", exporter);
Whitebox.setInternalState(grpcExporterProvider, "grpcMetricsExporter", exporter);
grpcExporterProvider.notifyAfterCompleted();
}
@ -106,4 +108,4 @@ public class GRPCExporterProviderTest {
assertEquals(1, requireModules.length);
assertEquals("core", requireModules[0]);
}
}
}

12
oap-server/exporter/src/test/java/org/apache/skywalking/oap/server/exporter/provider/grpc/GRPCExporterTest.java
View File

@ -28,6 +28,7 @@ import org.apache.skywalking.oap.server.core.exporter.ExportData;
import org.apache.skywalking.oap.server.core.exporter.ExportEvent;
import org.apache.skywalking.oap.server.core.source.DefaultScopeDefine;
import org.apache.skywalking.oap.server.exporter.grpc.MetricExportServiceGrpc;
import org.apache.skywalking.oap.server.exporter.provider.ExporterSetting;
import org.junit.Before;
import org.junit.Rule;
import org.junit.Test;
@ -37,7 +38,7 @@ import static org.apache.skywalking.oap.server.core.exporter.ExportEvent.EventTy
public class GRPCExporterTest {
private GRPCExporter exporter;
private GRPCMetricsExporter exporter;
@Rule
public final GrpcServerRule grpcServerRule = new GrpcServerRule().directExecutor();
@ -49,13 +50,14 @@ public class GRPCExporterTest {
@Before
public void setUp() throws Exception {
GRPCExporterSetting setting = new GRPCExporterSetting();
setting.setTargetHost("localhost");
setting.setTargetPort(9870);
exporter = new GRPCExporter(setting);
ExporterSetting setting = new ExporterSetting();
setting.setGRPCTargetHost("localhost");
setting.setGRPCTargetPort(9870);
exporter = new GRPCMetricsExporter(setting);
grpcServerRule.getServiceRegistry().addService(service);
stub = MetricExportServiceGrpc.newBlockingStub(grpcServerRule.getChannel());
Whitebox.setInternalState(exporter, "blockingStub", stub);
exporter.start();
}
@Test

3
oap-server/pom.xml
View File

@ -51,9 +51,6 @@
<properties>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
<kafka-clients.version>2.4.1</kafka-clients.version>
<spring-kafka-test.version>2.4.6.RELEASE</spring-kafka-test.version>
</properties>
<dependencies>

9
oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/worker/ExportWorker.java → oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/worker/ExportMetricsWorker.java
View File

@ -28,10 +28,10 @@ import org.apache.skywalking.oap.server.library.module.ModuleDefineHolder;
* A bridge worker. If the {@link ExporterModule} provider declared and provides a implementation of {@link
* MetricValuesExportService}, forward the export data to it.
*/
public class ExportWorker extends AbstractWorker<ExportEvent> {
public class ExportMetricsWorker extends AbstractWorker<ExportEvent> {
private MetricValuesExportService exportService;
public ExportWorker(ModuleDefineHolder moduleDefineHolder) {
public ExportMetricsWorker(ModuleDefineHolder moduleDefineHolder) {
super(moduleDefineHolder);
}
@ -43,8 +43,9 @@ public class ExportWorker extends AbstractWorker<ExportEvent> {
.provider()
.getService(MetricValuesExportService.class);
}
exportService.export(event);
if (exportService.isEnabled()) {
exportService.export(event);
}
}
}
}

64
oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/worker/ExportRecordWorker.java Normal file
View File

@ -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.
*
*/
package org.apache.skywalking.oap.server.core.analysis.worker;
import org.apache.skywalking.oap.server.core.analysis.manual.log.LogRecord;
import org.apache.skywalking.oap.server.core.analysis.manual.segment.SegmentRecord;
import org.apache.skywalking.oap.server.core.analysis.record.Record;
import org.apache.skywalking.oap.server.core.exporter.ExporterModule;
import org.apache.skywalking.oap.server.core.exporter.LogExportService;
import org.apache.skywalking.oap.server.core.exporter.TraceExportService;
import org.apache.skywalking.oap.server.core.worker.AbstractWorker;
import org.apache.skywalking.oap.server.library.module.ModuleDefineHolder;
public class ExportRecordWorker extends AbstractWorker<Record> {
private TraceExportService traceExportService;
private LogExportService logExportService;
public ExportRecordWorker(ModuleDefineHolder moduleDefineHolder) {
super(moduleDefineHolder);
}
@Override
public void in(Record record) {
if (record instanceof SegmentRecord) {
if (traceExportService != null || getModuleDefineHolder().has(ExporterModule.NAME)) {
if (traceExportService == null) {
traceExportService = getModuleDefineHolder().find(ExporterModule.NAME)
.provider()
.getService(TraceExportService.class);
}
if (traceExportService.isEnabled()) {
traceExportService.export((SegmentRecord) record);
}
}
} else if (record instanceof LogRecord) {
if (logExportService != null || getModuleDefineHolder().has(ExporterModule.NAME)) {
if (logExportService == null) {
logExportService = getModuleDefineHolder().find(ExporterModule.NAME)
.provider()
.getService(LogExportService.class);
}
if (logExportService.isEnabled()) {
logExportService.export((LogRecord) record);
}
}
}
}
}

2
oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/worker/MetricsStreamProcessor.java
View File

@ -205,7 +205,7 @@ public class MetricsStreamProcessor implements StreamProcessor<Metrics> {
MetricsTransWorker transWorker,
boolean supportUpdate) {
AlarmNotifyWorker alarmNotifyWorker = new AlarmNotifyWorker(moduleDefineHolder);
ExportWorker exportWorker = new ExportWorker(moduleDefineHolder);
ExportMetricsWorker exportWorker = new ExportMetricsWorker(moduleDefineHolder);
MetricsPersistentWorker minutePersistentWorker = new MetricsPersistentWorker(
moduleDefineHolder, model, metricsDAO, alarmNotifyWorker, exportWorker, transWorker,

6
oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/worker/RecordPersistentWorker.java
View File

@ -19,6 +19,7 @@
package org.apache.skywalking.oap.server.core.analysis.worker;
import java.io.IOException;
import java.util.Optional;
import org.apache.skywalking.oap.server.core.analysis.record.Record;
import org.apache.skywalking.oap.server.core.storage.IBatchDAO;
import org.apache.skywalking.oap.server.core.storage.IRecordDAO;
@ -37,12 +38,14 @@ public class RecordPersistentWorker extends AbstractWorker<Record> {
private final Model model;
private final IRecordDAO recordDAO;
private final IBatchDAO batchDAO;
private final Optional<AbstractWorker<Record>> nextExportWorker;
RecordPersistentWorker(ModuleDefineHolder moduleDefineHolder, Model model, IRecordDAO recordDAO) {
RecordPersistentWorker(ModuleDefineHolder moduleDefineHolder, Model model, IRecordDAO recordDAO, AbstractWorker<Record> nextExportWorker) {
super(moduleDefineHolder);
this.model = model;
this.recordDAO = recordDAO;
this.batchDAO = moduleDefineHolder.find(StorageModule.NAME).provider().getService(IBatchDAO.class);
this.nextExportWorker = Optional.ofNullable(nextExportWorker);
}
@Override
@ -53,5 +56,6 @@ public class RecordPersistentWorker extends AbstractWorker<Record> {
} catch (IOException e) {
LOGGER.error(e.getMessage(), e);
}
this.nextExportWorker.ifPresent(exportWorker -> exportWorker.in(record));
}
}

3
oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/worker/RecordStreamProcessor.java
View File

@ -76,7 +76,8 @@ public class RecordStreamProcessor implements StreamProcessor<Record> {
// Record stream doesn't read data from database during the persistent process. Keep the timeRelativeID == false always.
Model model = modelSetter.add(
recordClass, stream.scopeId(), new Storage(stream.name(), false, DownSampling.Second), true);
RecordPersistentWorker persistentWorker = new RecordPersistentWorker(moduleDefineHolder, model, recordDAO);
ExportRecordWorker exportWorker = new ExportRecordWorker(moduleDefineHolder);
RecordPersistentWorker persistentWorker = new RecordPersistentWorker(moduleDefineHolder, model, recordDAO, exportWorker);
workers.put(recordClass, persistentWorker);
}

5
oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/exporter/ExporterModule.java
View File

@ -29,6 +29,9 @@ public class ExporterModule extends ModuleDefine {
@Override
public Class[] services() {
return new Class[] {MetricValuesExportService.class};
return new Class[] {
MetricValuesExportService.class,
TraceExportService.class,
LogExportService.class};
}
}

20
oap-server/exporter/src/main/java/org/apache/skywalking/oap/server/exporter/provider/grpc/GRPCExporterSetting.java → oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/exporter/ExporterService.java
View File

@ -16,17 +16,15 @@
*
*/
package org.apache.skywalking.oap.server.exporter.provider.grpc;
package org.apache.skywalking.oap.server.core.exporter;
import lombok.Getter;
import lombok.Setter;
import org.apache.skywalking.oap.server.library.module.ModuleConfig;
import org.apache.skywalking.oap.server.library.module.Service;
@Setter
@Getter
public class GRPCExporterSetting extends ModuleConfig {
private String targetHost;
private int targetPort;
private int bufferChannelSize = 20000;
private int bufferChannelNum = 2;
public interface ExporterService<T> extends Service {
void start();
void export(T data);
boolean isEnabled();
}

30
oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/exporter/LogExportService.java Normal file
View File

@ -0,0 +1,30 @@
/*
* 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 org.apache.skywalking.oap.server.core.exporter;
import org.apache.skywalking.oap.server.core.analysis.manual.log.LogRecord;
import org.apache.skywalking.oap.server.library.module.Service;
/**
* Export the log from metrics through this service.
*/
public interface LogExportService extends Service, ExporterService<LogRecord> {
void export(LogRecord logRecord);
}

4
oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/exporter/MetricValuesExportService.java
View File

@ -21,9 +21,9 @@ package org.apache.skywalking.oap.server.core.exporter;
import org.apache.skywalking.oap.server.library.module.Service;
/**
* Export the metrics value from metrics through this service, if provider exists.
* Export the metrics value from metrics through this service
*/
public interface MetricValuesExportService extends Service {
public interface MetricValuesExportService extends Service, ExporterService<ExportEvent> {
/**
* This method is sync-mode export, the performance effects the persistence result. Queue mode is highly
* recommended.

30
oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/exporter/TraceExportService.java Normal file
View File

@ -0,0 +1,30 @@
/*
* 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 org.apache.skywalking.oap.server.core.exporter;
import org.apache.skywalking.oap.server.core.analysis.manual.segment.SegmentRecord;
import org.apache.skywalking.oap.server.library.module.Service;
/**
* Export the traces from metrics through this service.
*/
public interface TraceExportService extends Service, ExporterService<SegmentRecord> {
void export(SegmentRecord segmentRecord);
}

5
oap-server/server-fetcher-plugin/kafka-fetcher-plugin/pom.xml
View File

@ -42,13 +42,10 @@
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka-clients</artifactId>
<version>${kafka-clients.version}</version>
</dependency>
<dependency>
<groupId>org.springframework.kafka</groupId>
<artifactId>spring-kafka-test</artifactId>
<version>${spring-kafka-test.version}</version>
<scope>test</scope>
</dependency>
</dependencies>
</project>
</project>

1
oap-server/server-receiver-plugin/zipkin-receiver-plugin/pom.xml
View File

@ -36,7 +36,6 @@
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka-clients</artifactId>
<version>${kafka-clients.version}</version>
</dependency>
</dependencies>
</project>

16
oap-server/server-starter/src/main/resources/application.yml
View File

@ -501,9 +501,19 @@ configuration:
exporter:
selector: ${SW_EXPORTER:-}
grpc:
targetHost: ${SW_EXPORTER_GRPC_HOST:127.0.0.1}
targetPort: ${SW_EXPORTER_GRPC_PORT:9870}
default:
# gRPC exporter
enableGRPCMetrics: ${SW_EXPORTER_ENABLE_GRPC_METRICS:false}
gRPCTargetHost: ${SW_EXPORTER_GRPC_HOST:127.0.0.1}
gRPCTargetPort: ${SW_EXPORTER_GRPC_PORT:9870}
# Kafka exporter
enableKafkaTrace: ${SW_EXPORTER_ENABLE_KAFKA_TRACE:false}
enableKafkaLog: ${SW_EXPORTER_ENABLE_KAFKA_LOG:false}
kafkaBootstrapServers: ${SW_EXPORTER_KAFKA_SERVERS:localhost:9092}
# Kafka producer config, JSON format as Properties.
kafkaProducerConfig: ${SW_EXPORTER_KAFKA_PRODUCER_CONFIG:""}
kafkaTopicTrace: ${SW_EXPORTER_KAFKA_TOPIC_TRACE:skywalking-export-trace}
kafkaTopicLog: ${SW_EXPORTER_KAFKA_TOPIC_LOG:skywalking-export-log}
health-checker:
selector: ${SW_HEALTH_CHECKER:-}

93
test/e2e-v2/cases/exporter/kafka/docker-compose.yml Normal file
View File

@ -0,0 +1,93 @@
# 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.4
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
oap:
extends:
file: ../../../script/docker-compose/base-compose.yml
service: oap
environment:
SW_EXPORTER: default
SW_EXPORTER_ENABLE_KAFKA_TRACE: "true"
SW_EXPORTER_ENABLE_KAFKA_LOG: "true"
SW_EXPORTER_KAFKA_SERVERS: broker-a:9092
SW_TELEMETRY: prometheus
ports:
- 1234
- 12800
depends_on:
broker-a:
condition: service_healthy
networks:
- e2e
provider:
extends:
file: ../../../script/docker-compose/base-compose.yml
service: provider
ports:
- 9090
depends_on:
oap:
condition: service_healthy
consumer:
extends:
file: ../../../script/docker-compose/base-compose.yml
service: consumer
ports:
- 9092
depends_on:
provider:
condition: service_healthy
networks:
e2e:

47
test/e2e-v2/cases/exporter/kafka/e2e.yaml Normal file
View File

@ -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.
# 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: 20m
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-v2/script/prepare/setup-e2e-shell/install.sh yq
- name: install swctl
command: bash test/e2e-v2/script/prepare/setup-e2e-shell/install.sh swctl
trigger:
action: http
interval: 3s
times: 10
url: http://${consumer_host}:${consumer_9092}/users
method: POST
body: '{"id":"123","name":"skywalking"}'
headers:
"Content-Type": "application/json"
verify:
retry:
count: 20
interval: 3s
cases:
- includes:
- exporter-cases.yaml

17
test/e2e-v2/cases/exporter/kafka/expected/result.yml Normal file
View File

@ -0,0 +1,17 @@
# 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.
# "1" represents all received traces/logs were exported successfully: {exported / received == 1 }
result: 1

34
test/e2e-v2/cases/exporter/kafka/exporter-cases.yaml Normal file
View File

@ -0,0 +1,34 @@
# 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.
cases:
# verify traces
- query: |
sleep 10;
trace_in=$(curl -s http://${oap_host}:${oap_1234} | grep trace_in_latency_count{ | awk '{print $2}'); \
trace_out=$(curl -s http://${oap_host}:${oap_1234} | grep kafka_exporter_trace_success_count{ | awk '{print $2}'); \
awk 'BEGIN {print '$trace_out/$trace_in'}' | yq e '{"result": .}' -
expected: expected/result.yml
# verify logs
- query: |
curl -X POST http://${provider_host}:${provider_9090}/logs/trigger > /dev/null;
curl -X POST http://${provider_host}:${provider_9090}/logs/trigger > /dev/null;
curl -X POST http://${provider_host}:${provider_9090}/logs/trigger > /dev/null;
sleep 10;
log_in=$(curl -s http://${oap_host}:${oap_1234} | grep log_in_latency_count{ | awk '{print $2}'); \
log_out=$(curl -s http://${oap_host}:${oap_1234} | grep kafka_exporter_log_success_count{ | awk '{print $2}'); \
awk 'BEGIN {print '$log_out/$log_in'}' | yq e '{"result": .}' -
expected: expected/result.yml