You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@eventmesh.apache.org by mi...@apache.org on 2022/02/24 08:11:25 UTC
[incubator-eventmesh] branch master updated: [ISSUE #696] Add trace plugin (#749)
This is an automated email from the ASF dual-hosted git repository.
mikexue pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-eventmesh.git
The following commit(s) were added to refs/heads/master by this push:
new cde95c0 [ISSUE #696] Add trace plugin (#749)
cde95c0 is described below
commit cde95c0ab7630410ce672b2c319e0c7a9efd040c
Author: ZePeng Chen <84...@users.noreply.github.com>
AuthorDate: Thu Feb 24 16:11:18 2022 +0800
[ISSUE #696] Add trace plugin (#749)
* add docs
* add trace plugin
* fix ConfigurationWrapperTest error
* fix checkstyle
* fix checkstyle
---
.../common/config/CommonConfiguration.java | 78 ++------------
.../src/test/resources/configuration.properties | 3 +-
eventmesh-runtime/build.gradle | 3 +
eventmesh-runtime/conf/eventmesh.properties | 17 +--
.../eventmesh/runtime/boot/AbstractHTTPServer.java | 5 +-
.../runtime/boot/EventMeshHTTPServer.java | 24 ++++-
.../eventmesh/runtime/exporter/ZipkinExporter.java | 46 --------
.../runtime/{exporter => trace}/LogExporter.java | 2 +-
.../runtime/trace/OpenTelemetryTraceFactory.java | 109 -------------------
.../eventmesh/spi/EventMeshExtensionType.java | 1 +
.../build.gradle | 15 +--
.../eventmesh-trace-api/build.gradle | 19 ++--
.../eventmesh-trace-api/gradle.properties | 13 +--
.../eventmesh/trace/api/TracePluginFactory.java | 39 +++----
.../apache/eventmesh/trace/api/TraceService.java | 49 +++++----
.../trace/api/config/ExporterConfiguration.java | 117 ++++++++++++++++++++
.../src/main/resources/exporter.properties | 20 ++--
.../eventmesh-trace-zipkin/build.gradle | 22 ++--
.../eventmesh-trace-zipkin/gradle.properties | 13 +--
.../eventmesh/trace/zipkin/ZipkinTraceService.java | 118 +++++++++++++++++++++
.../trace/zipkin/config/ZipkinConfiguration.java | 96 +++++++++++++++++
.../org.apache.eventmesh.trace.api.TraceService | 15 +--
.../src/main/resources/zipkin.properties | 15 +--
.../gradle.properties | 13 +--
settings.gradle | 3 +
25 files changed, 465 insertions(+), 390 deletions(-)
diff --git a/eventmesh-common/src/main/java/org/apache/eventmesh/common/config/CommonConfiguration.java b/eventmesh-common/src/main/java/org/apache/eventmesh/common/config/CommonConfiguration.java
index a20d427..8d3e3eb 100644
--- a/eventmesh-common/src/main/java/org/apache/eventmesh/common/config/CommonConfiguration.java
+++ b/eventmesh-common/src/main/java/org/apache/eventmesh/common/config/CommonConfiguration.java
@@ -38,13 +38,7 @@ public class CommonConfiguration {
public String eventMeshRegistryPluginType = "namesrv";
public List<String> eventMeshMetricsPluginType;
- public String eventMeshTraceExporterType = "Log";
- public int eventMeshTraceExporterMaxExportSize = 512;
- public int eventMeshTraceExporterMaxQueueSize = 2048;
- public int eventMeshTraceExporterExportTimeout = 30;
- public int eventMeshTraceExporterExportInterval = 5;
- public String eventMeshTraceExportZipkinIp = "localhost";
- public int eventMeshTraceExportZipkinPort = 9411;
+ public String eventMeshTracePluginType;
public String namesrvAddr = "";
public Integer eventMeshRegisterIntervalInMills = 10 * 1000;
@@ -126,54 +120,6 @@ public class CommonConfiguration {
Preconditions.checkState(StringUtils.isNotEmpty(eventMeshRegistryPluginType),
String.format("%s error", ConfKeys.KEYS_ENENTMESH_REGISTRY_PLUGIN_TYPE));
- String eventMeshTraceExporterTypeStr =
- configurationWrapper.getProp(ConfKeys.KEYS_ENENTMESH_TRACE_EXPORTER_TYPE);
- Preconditions.checkState(StringUtils.isNotEmpty(eventMeshTraceExporterTypeStr),
- String.format("%s error", ConfKeys.KEYS_ENENTMESH_TRACE_EXPORTER_TYPE));
- eventMeshTraceExporterType =
- StringUtils.deleteWhitespace(eventMeshTraceExporterTypeStr);
-
- String eventMeshTraceExporterMaxExportSizeStr =
- configurationWrapper.getProp(ConfKeys.KEYS_EVENTMESH_TRACE_EXPORTER_MAX_EXPORT_SIZE);
- if (StringUtils.isNotEmpty(eventMeshTraceExporterMaxExportSizeStr)) {
- eventMeshTraceExporterMaxExportSize =
- Integer.parseInt(StringUtils.deleteWhitespace(eventMeshTraceExporterMaxExportSizeStr));
- }
-
- String eventMeshTraceExporterMaxQueueSizeStr =
- configurationWrapper.getProp(ConfKeys.KEYS_EVENTMESH_TRACE_EXPORTER_MAX_QUEUE_SIZE);
- if (StringUtils.isNotEmpty(eventMeshTraceExporterMaxQueueSizeStr)) {
- eventMeshTraceExporterMaxQueueSize =
- Integer.parseInt(StringUtils.deleteWhitespace(eventMeshTraceExporterMaxQueueSizeStr));
- }
-
- String eventMeshTraceExporterExportTimeoutStr =
- configurationWrapper.getProp(ConfKeys.KEYS_EVENTMESH_TRACE_EXPORTER_EXPORT_TIMEOUT);
- if (StringUtils.isNotEmpty(eventMeshTraceExporterExportTimeoutStr)) {
- eventMeshTraceExporterExportTimeout =
- Integer.parseInt(StringUtils.deleteWhitespace(eventMeshTraceExporterExportTimeoutStr));
- }
-
- String eventMeshTraceExporterExportIntervalStr =
- configurationWrapper.getProp(ConfKeys.KEYS_EVENTMESH_TRACE_EXPORTER_EXPORT_INTERVAL);
- if (StringUtils.isNotEmpty(eventMeshTraceExporterExportIntervalStr)) {
- eventMeshTraceExporterExportInterval =
- Integer.parseInt(StringUtils.deleteWhitespace(eventMeshTraceExporterExportIntervalStr));
- }
-
- String eventMeshTraceExportZipkinIpStr =
- configurationWrapper.getProp(ConfKeys.KEYS_EVENTMESH_TRACE_EXPORT_ZIPKIN_IP);
- Preconditions.checkState(StringUtils.isNotEmpty(eventMeshTraceExportZipkinIpStr),
- String.format("%s error", ConfKeys.KEYS_EVENTMESH_TRACE_EXPORT_ZIPKIN_IP));
- eventMeshTraceExportZipkinIp = StringUtils.deleteWhitespace(eventMeshTraceExportZipkinIpStr);
-
- String eventMeshTraceExportZipkinPortStr =
- configurationWrapper.getProp(ConfKeys.KEYS_EVENTMESH_TRACE_EXPORT_ZIPKIN_PORT);
- if (StringUtils.isNotEmpty(eventMeshTraceExportZipkinPortStr)) {
- eventMeshTraceExportZipkinPort =
- Integer.parseInt(StringUtils.deleteWhitespace(eventMeshTraceExportZipkinPortStr));
- }
-
String metricsPluginType = configurationWrapper.getProp(ConfKeys.KEYS_EVENTMESH_METRICS_PLUGIN_TYPE);
if (StringUtils.isNotEmpty(metricsPluginType)) {
eventMeshMetricsPluginType = Arrays.stream(metricsPluginType.split(","))
@@ -181,6 +127,12 @@ public class CommonConfiguration {
.map(String::trim)
.collect(Collectors.toList());
}
+
+ String eventMeshTracePluginTypeStr =
+ configurationWrapper.getProp(ConfKeys.KEYS_EVENTMESH_TRACE_PLUGIN_TYPE);
+ Preconditions.checkState(StringUtils.isNotEmpty(eventMeshTracePluginTypeStr),
+ String.format("%s error", ConfKeys.KEYS_EVENTMESH_TRACE_PLUGIN_TYPE));
+ eventMeshTracePluginType = StringUtils.deleteWhitespace(eventMeshTracePluginTypeStr);
}
}
@@ -213,20 +165,8 @@ public class CommonConfiguration {
public static String KEYS_ENENTMESH_REGISTRY_PLUGIN_TYPE = "eventMesh.registry.plugin.type";
- public static String KEYS_ENENTMESH_TRACE_EXPORTER_TYPE = "eventmesh.trace.exporter.type";
-
- public static String KEYS_EVENTMESH_TRACE_EXPORTER_MAX_EXPORT_SIZE = "eventmesh.trace.exporter.max.export.size";
-
- public static String KEYS_EVENTMESH_TRACE_EXPORTER_MAX_QUEUE_SIZE = "eventmesh.trace.exporter.max.queue.size";
-
- public static String KEYS_EVENTMESH_TRACE_EXPORTER_EXPORT_TIMEOUT = "eventmesh.trace.exporter.export.timeout";
-
- public static String KEYS_EVENTMESH_TRACE_EXPORTER_EXPORT_INTERVAL = "eventmesh.trace.exporter.export.interval";
-
- public static String KEYS_EVENTMESH_TRACE_EXPORT_ZIPKIN_IP = "eventmesh.trace.export.zipkin.ip";
-
- public static String KEYS_EVENTMESH_TRACE_EXPORT_ZIPKIN_PORT = "eventmesh.trace.export.zipkin.port";
-
public static String KEYS_EVENTMESH_METRICS_PLUGIN_TYPE = "eventmesh.metrics.plugin";
+
+ public static String KEYS_EVENTMESH_TRACE_PLUGIN_TYPE = "eventmesh.trace.plugin";
}
}
\ No newline at end of file
diff --git a/eventmesh-common/src/test/resources/configuration.properties b/eventmesh-common/src/test/resources/configuration.properties
index d4a7542..b04eab6 100644
--- a/eventmesh-common/src/test/resources/configuration.properties
+++ b/eventmesh-common/src/test/resources/configuration.properties
@@ -23,5 +23,4 @@ eventMesh.server.hostIp=value6
eventMesh.connector.plugin.type=rocketmq
eventMesh.security.plugin.type=acl
eventMesh.registry.plugin.type=namesrv
-eventmesh.trace.export.zipkin.ip=localhost
-eventmesh.trace.exporter.type=Zipkin
\ No newline at end of file
+eventmesh.trace.plugin=zipkin
\ No newline at end of file
diff --git a/eventmesh-runtime/build.gradle b/eventmesh-runtime/build.gradle
index 4c0e492..54e763d 100644
--- a/eventmesh-runtime/build.gradle
+++ b/eventmesh-runtime/build.gradle
@@ -57,6 +57,9 @@ dependencies {
implementation project(":eventmesh-metrics-plugin:eventmesh-metrics-api")
implementation project(":eventmesh-metrics-plugin:eventmesh-metrics-opentelemetry")
+ implementation project(":eventmesh-trace-plugin:eventmesh-trace-api")
+ implementation project(":eventmesh-trace-plugin:eventmesh-trace-zipkin")
+
testImplementation "org.mockito:mockito-core"
testImplementation "org.powermock:powermock-module-junit4"
testImplementation "org.powermock:powermock-api-mockito2"
diff --git a/eventmesh-runtime/conf/eventmesh.properties b/eventmesh-runtime/conf/eventmesh.properties
index 485886c..538cc67 100644
--- a/eventmesh-runtime/conf/eventmesh.properties
+++ b/eventmesh-runtime/conf/eventmesh.properties
@@ -84,18 +84,5 @@ eventMesh.registry.plugin.type=namesrv
# metrics plugin, if you have multiple plugin, you can use ',' to split
eventmesh.metrics.plugin=opentelemetry
-#trace exporter
-eventmesh.trace.exporter.type=Log
-
-#set the maximum batch size to use
-eventmesh.trace.exporter.max.export.size=512
-#set the queue size. This must be >= the export batch size
-eventmesh.trace.exporter.max.queue.size=2048
-#set the max amount of time an export can run before getting(TimeUnit=SECONDS)
-eventmesh.trace.exporter.export.timeout=30
-#set time between two different exports(TimeUnit=SECONDS)
-eventmesh.trace.exporter.export.interval=5
-
-#zipkin
-eventmesh.trace.export.zipkin.ip=localhost
-eventmesh.trace.export.zipkin.port=9411
\ No newline at end of file
+# trace plugin
+eventmesh.trace.plugin=zipkin
\ No newline at end of file
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/AbstractHTTPServer.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/AbstractHTTPServer.java
index 592854d..7af7c36 100644
--- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/AbstractHTTPServer.java
+++ b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/AbstractHTTPServer.java
@@ -31,7 +31,6 @@ import org.apache.eventmesh.runtime.core.protocol.http.async.AsyncContext;
import org.apache.eventmesh.runtime.core.protocol.http.processor.inf.HttpRequestProcessor;
import org.apache.eventmesh.runtime.metrics.http.HTTPMetricsServer;
import org.apache.eventmesh.runtime.trace.AttributeKeys;
-import org.apache.eventmesh.runtime.trace.OpenTelemetryTraceFactory;
import org.apache.eventmesh.runtime.trace.SpanKey;
import org.apache.eventmesh.runtime.util.RemotingHelper;
@@ -111,12 +110,10 @@ public abstract class AbstractHTTPServer extends AbstractRemotingServer {
private boolean useTLS;
- private Boolean useTrace = true; //Determine whether trace is enabled
+ public Boolean useTrace = false; //Determine whether trace is enabled
public TextMapPropagator textMapPropagator;
- public OpenTelemetryTraceFactory openTelemetryTraceFactory;
-
public Tracer tracer;
public ThreadPoolExecutor asyncContextCompleteHandler =
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/EventMeshHTTPServer.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/EventMeshHTTPServer.java
index 590138a..dfa6e6b 100644
--- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/EventMeshHTTPServer.java
+++ b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/EventMeshHTTPServer.java
@@ -39,7 +39,10 @@ import org.apache.eventmesh.runtime.core.protocol.http.producer.ProducerManager;
import org.apache.eventmesh.runtime.core.protocol.http.push.AbstractHTTPPushRequest;
import org.apache.eventmesh.runtime.core.protocol.http.retry.HttpRetryer;
import org.apache.eventmesh.runtime.metrics.http.HTTPMetricsServer;
-import org.apache.eventmesh.runtime.trace.OpenTelemetryTraceFactory;
+import org.apache.eventmesh.trace.api.TracePluginFactory;
+import org.apache.eventmesh.trace.api.TraceService;
+
+import org.apache.commons.lang3.StringUtils;
import java.util.List;
import java.util.Optional;
@@ -61,6 +64,8 @@ public class EventMeshHTTPServer extends AbstractHTTPServer {
private EventMeshHTTPConfiguration eventMeshHttpConfiguration;
+ private TraceService traceService;
+
public final ConcurrentHashMap<String /**group*/, ConsumerGroupConf> localConsumerGroupMapping =
new ConcurrentHashMap<>();
@@ -216,9 +221,16 @@ public class EventMeshHTTPServer extends AbstractHTTPServer {
registerHTTPRequestProcessor();
- super.openTelemetryTraceFactory = new OpenTelemetryTraceFactory(eventMeshHttpConfiguration);
- super.tracer = openTelemetryTraceFactory.getTracer(this.getClass().toString());
- super.textMapPropagator = openTelemetryTraceFactory.getTextMapPropagator();
+ //get the trace-plugin
+ if (StringUtils.isNotEmpty(eventMeshHttpConfiguration.eventMeshTracePluginType)) {
+
+ traceService =
+ TracePluginFactory.getTraceService(eventMeshHttpConfiguration.eventMeshTracePluginType);
+ traceService.init();
+ super.tracer = traceService.getTracer(super.getClass().toString());
+ super.textMapPropagator = traceService.getTextMapPropagator();
+ super.useTrace = true;
+ }
logger.info("--------------------------EventMeshHTTPServer inited");
}
@@ -240,6 +252,10 @@ public class EventMeshHTTPServer extends AbstractHTTPServer {
metrics.shutdown();
+ if (traceService != null) {
+ traceService.shutdown();
+ }
+
consumerManager.shutdown();
shutdownThreadPool();
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/exporter/ZipkinExporter.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/exporter/ZipkinExporter.java
deleted file mode 100644
index c5b0284..0000000
--- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/exporter/ZipkinExporter.java
+++ /dev/null
@@ -1,46 +0,0 @@
-/*
- * 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.eventmesh.runtime.exporter;
-
-import org.apache.eventmesh.common.config.CommonConfiguration;
-
-import io.opentelemetry.exporter.zipkin.ZipkinSpanExporter;
-
-/**
- * an exporter to export traced data to zipkin.
- */
-public class ZipkinExporter implements EventMeshExporter {
- private String ip = "localhost";
-
- private int port = 9411;
-
- // Zipkin API Endpoints for uploading spans
- private static final String ENDPOINT_V2_SPANS = "/api/v2/spans";
-
- private ZipkinSpanExporter zipkinExporter;
-
- @Override
- public ZipkinSpanExporter getSpanExporter(CommonConfiguration configuration) {
- ip = configuration.eventMeshServerIp;
- port = configuration.eventMeshTraceExportZipkinPort;
- String httpUrl = String.format("http://%s:%s", ip, port);
- zipkinExporter =
- ZipkinSpanExporter.builder().setEndpoint(httpUrl + ENDPOINT_V2_SPANS).build();
- return zipkinExporter;
- }
-}
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/exporter/LogExporter.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/trace/LogExporter.java
similarity index 98%
rename from eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/exporter/LogExporter.java
rename to eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/trace/LogExporter.java
index b4b11a4..9bad687 100644
--- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/exporter/LogExporter.java
+++ b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/trace/LogExporter.java
@@ -15,7 +15,7 @@
* limitations under the License.
*/
-package org.apache.eventmesh.runtime.exporter;
+package org.apache.eventmesh.runtime.trace;
import java.util.Collection;
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/trace/OpenTelemetryTraceFactory.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/trace/OpenTelemetryTraceFactory.java
deleted file mode 100644
index ae4ce41..0000000
--- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/trace/OpenTelemetryTraceFactory.java
+++ /dev/null
@@ -1,109 +0,0 @@
-/*
- * 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.eventmesh.runtime.trace;
-
-import static io.opentelemetry.api.common.AttributeKey.stringKey;
-
-import org.apache.eventmesh.common.config.CommonConfiguration;
-import org.apache.eventmesh.runtime.exporter.EventMeshExporter;
-import org.apache.eventmesh.runtime.exporter.LogExporter;
-
-import java.util.concurrent.TimeUnit;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import io.opentelemetry.api.OpenTelemetry;
-import io.opentelemetry.api.common.Attributes;
-import io.opentelemetry.api.trace.Tracer;
-import io.opentelemetry.api.trace.propagation.W3CTraceContextPropagator;
-import io.opentelemetry.context.propagation.ContextPropagators;
-import io.opentelemetry.context.propagation.TextMapPropagator;
-import io.opentelemetry.sdk.OpenTelemetrySdk;
-import io.opentelemetry.sdk.resources.Resource;
-import io.opentelemetry.sdk.trace.SdkTracerProvider;
-import io.opentelemetry.sdk.trace.SpanProcessor;
-import io.opentelemetry.sdk.trace.export.BatchSpanProcessor;
-import io.opentelemetry.sdk.trace.export.SpanExporter;
-
-/**
- * create tracer.
- */
-public class OpenTelemetryTraceFactory {
- private static final Logger logger = LoggerFactory.getLogger(OpenTelemetryTraceFactory.class);
- // Name of the service(using the instrumentationName)
- private final String serviceName = "eventmesh_trace";
- private OpenTelemetry openTelemetry;
- private SpanExporter spanExporter;
- private SpanExporter defaultExporter = new LogExporter();
- private SpanProcessor spanProcessor;
-
- public OpenTelemetryTraceFactory(CommonConfiguration configuration) {
- try {
- //different spanExporter
- String exporterName = configuration.eventMeshTraceExporterType;
- //use reflection to get spanExporter
- String className = String.format("org.apache.eventmesh.runtime.exporter.%sExporter", exporterName);
- EventMeshExporter eventMeshExporter = (EventMeshExporter) Class.forName(className).newInstance();
- spanExporter = eventMeshExporter.getSpanExporter(configuration);
- } catch (Exception ex) {
- logger.error("fail to set tracer's exporter,due to {}", ex.getMessage());
- //fail to set the exporter in configuration, changing to use the default Exporter
- spanExporter = defaultExporter;
- logger.info("change to use the default exporter {}", defaultExporter.getClass());
- }
-
- // Configure the batch spans processor. This span processor exports span in batches.
- spanProcessor = BatchSpanProcessor.builder(spanExporter)
- // set the maximum batch size to use
- .setMaxExportBatchSize(configuration.eventMeshTraceExporterMaxExportSize)
- // set the queue size. This must be >= the export batch size
- .setMaxQueueSize(configuration.eventMeshTraceExporterMaxQueueSize)
- // set the max amount of time an export can run before getting
- .setExporterTimeout(configuration.eventMeshTraceExporterExportTimeout, TimeUnit.SECONDS)
- // set time between two different exports
- .setScheduleDelay(configuration.eventMeshTraceExporterExportInterval, TimeUnit.SECONDS)
- .build();
-
- //set the trace service's name
- Resource serviceNameResource =
- Resource.create(Attributes.of(stringKey("service.name"), serviceName));
-
- SdkTracerProvider sdkTracerProvider = SdkTracerProvider.builder()
- .addSpanProcessor(spanProcessor)
- .setResource(Resource.getDefault().merge(serviceNameResource))
- .build();
-
- openTelemetry = OpenTelemetrySdk.builder()
- .setPropagators(ContextPropagators.create(W3CTraceContextPropagator.getInstance()))
- .setTracerProvider(sdkTracerProvider)
- .build();
-
- Runtime.getRuntime().addShutdownHook(new Thread(sdkTracerProvider::close));
- }
-
- //Gets or creates a named tracer instance
- public Tracer getTracer(String instrumentationName) {
- return openTelemetry.getTracer(instrumentationName);
- }
-
- //to inject or extract span context
- public TextMapPropagator getTextMapPropagator() {
- return openTelemetry.getPropagators().getTextMapPropagator();
- }
-}
diff --git a/eventmesh-spi/src/main/java/org/apache/eventmesh/spi/EventMeshExtensionType.java b/eventmesh-spi/src/main/java/org/apache/eventmesh/spi/EventMeshExtensionType.java
index e126bed..5aab9d1 100644
--- a/eventmesh-spi/src/main/java/org/apache/eventmesh/spi/EventMeshExtensionType.java
+++ b/eventmesh-spi/src/main/java/org/apache/eventmesh/spi/EventMeshExtensionType.java
@@ -27,6 +27,7 @@ public enum EventMeshExtensionType {
SECURITY("security"),
PROTOCOL("protocol"),
METRICS("metrics"),
+ TRACE("trace"),
;
private final String extensionTypeName;
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/exporter/EventMeshExporter.java b/eventmesh-trace-plugin/build.gradle
similarity index 69%
copy from eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/exporter/EventMeshExporter.java
copy to eventmesh-trace-plugin/build.gradle
index abc9cf9..d973dce 100644
--- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/exporter/EventMeshExporter.java
+++ b/eventmesh-trace-plugin/build.gradle
@@ -13,17 +13,4 @@
* 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.eventmesh.runtime.exporter;
-
-import org.apache.eventmesh.common.config.CommonConfiguration;
-
-import io.opentelemetry.sdk.trace.export.SpanExporter;
-
-/**
- * different exporters should implement this interface.
- */
-public interface EventMeshExporter {
- public SpanExporter getSpanExporter(CommonConfiguration configuration);
-}
+ */
\ No newline at end of file
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/exporter/EventMeshExporter.java b/eventmesh-trace-plugin/eventmesh-trace-api/build.gradle
similarity index 64%
copy from eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/exporter/EventMeshExporter.java
copy to eventmesh-trace-plugin/eventmesh-trace-api/build.gradle
index abc9cf9..36db8ec 100644
--- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/exporter/EventMeshExporter.java
+++ b/eventmesh-trace-plugin/eventmesh-trace-api/build.gradle
@@ -15,15 +15,16 @@
* limitations under the License.
*/
-package org.apache.eventmesh.runtime.exporter;
+dependencies {
+ api project(":eventmesh-spi")
+ implementation project(":eventmesh-common")
-import org.apache.eventmesh.common.config.CommonConfiguration;
+ implementation 'io.opentelemetry:opentelemetry-api'
+ implementation 'io.opentelemetry:opentelemetry-sdk'
-import io.opentelemetry.sdk.trace.export.SpanExporter;
+ compileOnly 'org.projectlombok:lombok:1.18.22'
+ annotationProcessor 'org.projectlombok:lombok:1.18.22'
-/**
- * different exporters should implement this interface.
- */
-public interface EventMeshExporter {
- public SpanExporter getSpanExporter(CommonConfiguration configuration);
-}
+ testCompileOnly 'org.projectlombok:lombok:1.18.22'
+ testAnnotationProcessor 'org.projectlombok:lombok:1.18.22'
+}
\ No newline at end of file
diff --git a/eventmesh-common/src/test/resources/configuration.properties b/eventmesh-trace-plugin/eventmesh-trace-api/gradle.properties
similarity index 68%
copy from eventmesh-common/src/test/resources/configuration.properties
copy to eventmesh-trace-plugin/eventmesh-trace-api/gradle.properties
index d4a7542..a9fd83f 100644
--- a/eventmesh-common/src/test/resources/configuration.properties
+++ b/eventmesh-trace-plugin/eventmesh-trace-api/gradle.properties
@@ -13,15 +13,4 @@
# 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.
-#
-eventMesh.server.env=value1
-eventMesh.server.idc=value2
-eventMesh.sysid=3
-eventMesh.server.cluster=value4
-eventMesh.server.name=value5
-eventMesh.server.hostIp=value6
-eventMesh.connector.plugin.type=rocketmq
-eventMesh.security.plugin.type=acl
-eventMesh.registry.plugin.type=namesrv
-eventmesh.trace.export.zipkin.ip=localhost
-eventmesh.trace.exporter.type=Zipkin
\ No newline at end of file
+#
\ No newline at end of file
diff --git a/eventmesh-spi/src/main/java/org/apache/eventmesh/spi/EventMeshExtensionType.java b/eventmesh-trace-plugin/eventmesh-trace-api/src/main/java/org/apache/eventmesh/trace/api/TracePluginFactory.java
similarity index 52%
copy from eventmesh-spi/src/main/java/org/apache/eventmesh/spi/EventMeshExtensionType.java
copy to eventmesh-trace-plugin/eventmesh-trace-api/src/main/java/org/apache/eventmesh/trace/api/TracePluginFactory.java
index e126bed..91e8d53 100644
--- a/eventmesh-spi/src/main/java/org/apache/eventmesh/spi/EventMeshExtensionType.java
+++ b/eventmesh-trace-plugin/eventmesh-trace-api/src/main/java/org/apache/eventmesh/trace/api/TracePluginFactory.java
@@ -15,28 +15,29 @@
* limitations under the License.
*/
-package org.apache.eventmesh.spi;
+package org.apache.eventmesh.trace.api;
-/**
- * An Extension can be defined by extensionTypeName and extensionInstanceName
- */
-public enum EventMeshExtensionType {
- UNKNOWN("unknown"),
- CONNECTOR("connector"),
- REGISTRY("registry"),
- SECURITY("security"),
- PROTOCOL("protocol"),
- METRICS("metrics"),
- ;
+import static com.google.common.base.Preconditions.checkNotNull;
- private final String extensionTypeName;
+import org.apache.eventmesh.spi.EventMeshExtensionFactory;
- EventMeshExtensionType(String extensionTypeName) {
- this.extensionTypeName = extensionTypeName;
- }
+import lombok.experimental.UtilityClass;
- public String getExtensionTypeName() {
- return extensionTypeName;
- }
+/**
+ * to get the trace service
+ */
+@UtilityClass
+public class TracePluginFactory {
+ /**
+ * to get TraceService
+ *
+ * @param traceServiceType
+ * @return
+ */
+ public static TraceService getTraceService(String traceServiceType) {
+ checkNotNull(traceServiceType, "traceServiceType cannot be null");
+ TraceService traceService = EventMeshExtensionFactory.getExtension(TraceService.class, traceServiceType);
+ return checkNotNull(traceService, "traceServiceType: " + traceServiceType + " is not supported");
+ }
}
diff --git a/eventmesh-spi/src/main/java/org/apache/eventmesh/spi/EventMeshExtensionType.java b/eventmesh-trace-plugin/eventmesh-trace-api/src/main/java/org/apache/eventmesh/trace/api/TraceService.java
similarity index 50%
copy from eventmesh-spi/src/main/java/org/apache/eventmesh/spi/EventMeshExtensionType.java
copy to eventmesh-trace-plugin/eventmesh-trace-api/src/main/java/org/apache/eventmesh/trace/api/TraceService.java
index e126bed..c98a3bb 100644
--- a/eventmesh-spi/src/main/java/org/apache/eventmesh/spi/EventMeshExtensionType.java
+++ b/eventmesh-trace-plugin/eventmesh-trace-api/src/main/java/org/apache/eventmesh/trace/api/TraceService.java
@@ -15,28 +15,41 @@
* limitations under the License.
*/
-package org.apache.eventmesh.spi;
+package org.apache.eventmesh.trace.api;
+
+import org.apache.eventmesh.spi.EventMeshExtensionType;
+import org.apache.eventmesh.spi.EventMeshSPI;
+
+import io.opentelemetry.api.trace.Tracer;
+import io.opentelemetry.context.propagation.TextMapPropagator;
/**
- * An Extension can be defined by extensionTypeName and extensionInstanceName
+ * The top-level interface of trace
*/
-public enum EventMeshExtensionType {
- UNKNOWN("unknown"),
- CONNECTOR("connector"),
- REGISTRY("registry"),
- SECURITY("security"),
- PROTOCOL("protocol"),
- METRICS("metrics"),
- ;
-
- private final String extensionTypeName;
+@EventMeshSPI(isSingleton = true, eventMeshExtensionType = EventMeshExtensionType.TRACE)
+public interface TraceService {
+ /**
+ * init the trace service
+ */
+ void init();
- EventMeshExtensionType(String extensionTypeName) {
- this.extensionTypeName = extensionTypeName;
- }
+ /**
+ * close the trace service
+ */
+ void shutdown();
- public String getExtensionTypeName() {
- return extensionTypeName;
- }
+ /**
+ * get the tracer
+ *
+ * @param instrumentationName
+ * @return
+ */
+ Tracer getTracer(String instrumentationName);
+ /**
+ * get TextMapPropagator
+ *
+ * @return
+ */
+ TextMapPropagator getTextMapPropagator();
}
diff --git a/eventmesh-trace-plugin/eventmesh-trace-api/src/main/java/org/apache/eventmesh/trace/api/config/ExporterConfiguration.java b/eventmesh-trace-plugin/eventmesh-trace-api/src/main/java/org/apache/eventmesh/trace/api/config/ExporterConfiguration.java
new file mode 100644
index 0000000..44d54d0
--- /dev/null
+++ b/eventmesh-trace-plugin/eventmesh-trace-api/src/main/java/org/apache/eventmesh/trace/api/config/ExporterConfiguration.java
@@ -0,0 +1,117 @@
+/*
+ * 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.eventmesh.trace.api.config;
+
+import org.apache.commons.lang3.StringUtils;
+
+import java.io.BufferedReader;
+import java.io.File;
+import java.io.FileReader;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.InputStreamReader;
+import java.net.URL;
+import java.util.Properties;
+
+import lombok.experimental.UtilityClass;
+import lombok.extern.slf4j.Slf4j;
+
+/**
+ * to load the properties form exporter.properties
+ */
+@Slf4j
+@UtilityClass
+public class ExporterConfiguration {
+
+ private static final String CONFIG_FILE = "exporter.properties";
+ private static final Properties properties = new Properties();
+
+ private int eventMeshTraceMaxExportSize = 512;
+ private int eventMeshTraceMaxQueueSize = 2048;
+ private int eventMeshTraceExportTimeout = 30;
+ private int eventMeshTraceExportInterval = 5;
+
+ static {
+ loadProperties();
+ initializeConfig();
+ }
+
+ public static int getEventMeshTraceMaxExportSize() {
+ return eventMeshTraceMaxExportSize;
+ }
+
+ public static int getEventMeshTraceMaxQueueSize() {
+ return eventMeshTraceMaxQueueSize;
+ }
+
+ public static int getEventMeshTraceExportTimeout() {
+ return eventMeshTraceExportTimeout;
+ }
+
+ public static int getEventMeshTraceExportInterval() {
+ return eventMeshTraceExportInterval;
+ }
+
+ private void initializeConfig() {
+ String eventMeshTraceMaxExportSizeStr = properties.getProperty("eventmesh.trace.max.export.size");
+ if (StringUtils.isNotEmpty(eventMeshTraceMaxExportSizeStr)) {
+ eventMeshTraceMaxExportSize =
+ Integer.parseInt(StringUtils.deleteWhitespace(eventMeshTraceMaxExportSizeStr));
+ }
+
+ String eventMeshTraceMaxQueueSizeStr = properties.getProperty("eventmesh.trace.max.queue.size");
+ if (StringUtils.isNotEmpty(eventMeshTraceMaxQueueSizeStr)) {
+ eventMeshTraceMaxQueueSize = Integer.parseInt(StringUtils.deleteWhitespace(eventMeshTraceMaxQueueSizeStr));
+ }
+
+ String eventMeshTraceExportTimeoutStr = properties.getProperty("eventmesh.trace.export.timeout");
+ if (StringUtils.isNotEmpty(eventMeshTraceExportTimeoutStr)) {
+ eventMeshTraceExportTimeout =
+ Integer.parseInt(StringUtils.deleteWhitespace(eventMeshTraceExportTimeoutStr));
+ }
+
+ String eventMeshTraceExportIntervalStr = properties.getProperty("eventmesh.trace.export.interval");
+ if (StringUtils.isNotEmpty(eventMeshTraceExportIntervalStr)) {
+ eventMeshTraceExportInterval =
+ Integer.parseInt(StringUtils.deleteWhitespace(eventMeshTraceExportIntervalStr));
+ }
+ }
+
+ private void loadProperties() {
+ URL resource = ExporterConfiguration.class.getClassLoader().getResource(CONFIG_FILE);
+ if (resource != null) {
+ try (InputStream inputStream = resource.openStream()) {
+ if (inputStream.available() > 0) {
+ properties.load(new BufferedReader(new InputStreamReader(inputStream)));
+ }
+ } catch (IOException e) {
+ throw new RuntimeException("Load exporter.properties file from classpath error");
+ }
+ }
+ // get from config home
+ try {
+ String configPath =
+ System.getProperty("confPath", System.getenv("confPath")) + File.separator + CONFIG_FILE;
+ if (new File(configPath).exists()) {
+ properties.load(new BufferedReader(new FileReader(configPath)));
+ }
+ } catch (IOException e) {
+ throw new IllegalArgumentException("Cannot load exporter.properties file from conf");
+ }
+ }
+}
diff --git a/eventmesh-common/src/test/resources/configuration.properties b/eventmesh-trace-plugin/eventmesh-trace-api/src/main/resources/exporter.properties
similarity index 67%
copy from eventmesh-common/src/test/resources/configuration.properties
copy to eventmesh-trace-plugin/eventmesh-trace-api/src/main/resources/exporter.properties
index d4a7542..a36d2d8 100644
--- a/eventmesh-common/src/test/resources/configuration.properties
+++ b/eventmesh-trace-plugin/eventmesh-trace-api/src/main/resources/exporter.properties
@@ -14,14 +14,12 @@
# See the License for the specific language governing permissions and
# limitations under the License.
#
-eventMesh.server.env=value1
-eventMesh.server.idc=value2
-eventMesh.sysid=3
-eventMesh.server.cluster=value4
-eventMesh.server.name=value5
-eventMesh.server.hostIp=value6
-eventMesh.connector.plugin.type=rocketmq
-eventMesh.security.plugin.type=acl
-eventMesh.registry.plugin.type=namesrv
-eventmesh.trace.export.zipkin.ip=localhost
-eventmesh.trace.exporter.type=Zipkin
\ No newline at end of file
+
+#set the maximum batch size to use
+eventmesh.trace.max.export.size=512
+#set the queue size. This must be >= the export batch size
+eventmesh.trace.max.queue.size=2048
+#set the max amount of time an export can run before getting(TimeUnit=SECONDS)
+eventmesh.trace.export.timeout=30
+#set time between two different exports(TimeUnit=SECONDS)
+eventmesh.trace.export.interval=5
\ No newline at end of file
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/exporter/EventMeshExporter.java b/eventmesh-trace-plugin/eventmesh-trace-zipkin/build.gradle
similarity index 56%
rename from eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/exporter/EventMeshExporter.java
rename to eventmesh-trace-plugin/eventmesh-trace-zipkin/build.gradle
index abc9cf9..124a9fd 100644
--- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/exporter/EventMeshExporter.java
+++ b/eventmesh-trace-plugin/eventmesh-trace-zipkin/build.gradle
@@ -15,15 +15,19 @@
* limitations under the License.
*/
-package org.apache.eventmesh.runtime.exporter;
+dependencies {
+ implementation project(":eventmesh-trace-plugin:eventmesh-trace-api")
+ implementation project(":eventmesh-common")
+ implementation 'org.slf4j:slf4j-api'
+ implementation 'org.apache.commons:commons-lang3'
+ implementation 'com.google.guava:guava'
-import org.apache.eventmesh.common.config.CommonConfiguration;
+ implementation 'io.opentelemetry:opentelemetry-exporter-zipkin'
+ implementation 'io.opentelemetry:opentelemetry-semconv'
-import io.opentelemetry.sdk.trace.export.SpanExporter;
+ compileOnly 'org.projectlombok:lombok:1.18.22'
+ annotationProcessor 'org.projectlombok:lombok:1.18.22'
-/**
- * different exporters should implement this interface.
- */
-public interface EventMeshExporter {
- public SpanExporter getSpanExporter(CommonConfiguration configuration);
-}
+ testCompileOnly 'org.projectlombok:lombok:1.18.22'
+ testAnnotationProcessor 'org.projectlombok:lombok:1.18.22'
+}
\ No newline at end of file
diff --git a/eventmesh-common/src/test/resources/configuration.properties b/eventmesh-trace-plugin/eventmesh-trace-zipkin/gradle.properties
similarity index 68%
copy from eventmesh-common/src/test/resources/configuration.properties
copy to eventmesh-trace-plugin/eventmesh-trace-zipkin/gradle.properties
index d4a7542..a9fd83f 100644
--- a/eventmesh-common/src/test/resources/configuration.properties
+++ b/eventmesh-trace-plugin/eventmesh-trace-zipkin/gradle.properties
@@ -13,15 +13,4 @@
# 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.
-#
-eventMesh.server.env=value1
-eventMesh.server.idc=value2
-eventMesh.sysid=3
-eventMesh.server.cluster=value4
-eventMesh.server.name=value5
-eventMesh.server.hostIp=value6
-eventMesh.connector.plugin.type=rocketmq
-eventMesh.security.plugin.type=acl
-eventMesh.registry.plugin.type=namesrv
-eventmesh.trace.export.zipkin.ip=localhost
-eventmesh.trace.exporter.type=Zipkin
\ No newline at end of file
+#
\ No newline at end of file
diff --git a/eventmesh-trace-plugin/eventmesh-trace-zipkin/src/main/java/org/apache/eventmesh/trace/zipkin/ZipkinTraceService.java b/eventmesh-trace-plugin/eventmesh-trace-zipkin/src/main/java/org/apache/eventmesh/trace/zipkin/ZipkinTraceService.java
new file mode 100644
index 0000000..4fdfcff
--- /dev/null
+++ b/eventmesh-trace-plugin/eventmesh-trace-zipkin/src/main/java/org/apache/eventmesh/trace/zipkin/ZipkinTraceService.java
@@ -0,0 +1,118 @@
+/*
+ * 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.eventmesh.trace.zipkin;
+
+import static io.opentelemetry.api.common.AttributeKey.stringKey;
+
+import org.apache.eventmesh.trace.api.TraceService;
+import org.apache.eventmesh.trace.api.config.ExporterConfiguration;
+import org.apache.eventmesh.trace.zipkin.config.ZipkinConfiguration;
+
+import java.util.concurrent.TimeUnit;
+
+import io.opentelemetry.api.OpenTelemetry;
+import io.opentelemetry.api.common.Attributes;
+import io.opentelemetry.api.trace.Tracer;
+import io.opentelemetry.api.trace.propagation.W3CTraceContextPropagator;
+import io.opentelemetry.context.propagation.ContextPropagators;
+import io.opentelemetry.context.propagation.TextMapPropagator;
+import io.opentelemetry.exporter.zipkin.ZipkinSpanExporter;
+import io.opentelemetry.sdk.OpenTelemetrySdk;
+import io.opentelemetry.sdk.resources.Resource;
+import io.opentelemetry.sdk.trace.SdkTracerProvider;
+import io.opentelemetry.sdk.trace.SpanProcessor;
+import io.opentelemetry.sdk.trace.export.BatchSpanProcessor;
+
+/**
+ *
+ */
+public class ZipkinTraceService implements TraceService {
+ // Zipkin API Endpoints for uploading spans
+ private static final String ENDPOINT_V2_SPANS = "/api/v2/spans";
+ // Name of the service(using the instrumentationName)
+ private final String serviceName = "eventmesh_trace";
+ private String eventMeshZipkinIP;
+ private int eventMeshZipkinPort;
+ private int eventMeshTraceExportInterval;
+ private int eventMeshTraceExportTimeout;
+ private int eventMeshTraceMaxExportSize;
+ private int eventMeshTraceMaxQueueSize;
+ private SdkTracerProvider sdkTracerProvider;
+
+ private OpenTelemetry openTelemetry;
+
+ @Override
+ public void init() {
+ //zipkin's config
+ eventMeshZipkinIP = ZipkinConfiguration.getEventMeshZipkinIP();
+ eventMeshZipkinPort = ZipkinConfiguration.getEventMeshZipkinPort();
+ //exporter's config
+ eventMeshTraceExportInterval = ExporterConfiguration.getEventMeshTraceExportInterval();
+ eventMeshTraceExportTimeout = ExporterConfiguration.getEventMeshTraceExportTimeout();
+ eventMeshTraceMaxExportSize = ExporterConfiguration.getEventMeshTraceMaxExportSize();
+ eventMeshTraceMaxQueueSize = ExporterConfiguration.getEventMeshTraceMaxQueueSize();
+
+ String httpUrl = String.format("http://%s:%s", eventMeshZipkinIP, eventMeshZipkinPort);
+ ZipkinSpanExporter zipkinExporter =
+ ZipkinSpanExporter.builder().setEndpoint(httpUrl + ENDPOINT_V2_SPANS).build();
+
+ SpanProcessor spanProcessor = BatchSpanProcessor.builder(zipkinExporter)
+ .setScheduleDelay(eventMeshTraceExportInterval, TimeUnit.SECONDS)
+ .setExporterTimeout(eventMeshTraceExportTimeout, TimeUnit.SECONDS)
+ .setMaxExportBatchSize(eventMeshTraceMaxExportSize)
+ .setMaxQueueSize(eventMeshTraceMaxQueueSize)
+ .build();
+
+ //set the trace service's name
+ Resource serviceNameResource =
+ Resource.create(Attributes.of(stringKey("service.name"), serviceName));
+
+ sdkTracerProvider = SdkTracerProvider.builder()
+ .addSpanProcessor(spanProcessor)
+ .setResource(Resource.getDefault().merge(serviceNameResource))
+ .build();
+
+ openTelemetry = OpenTelemetrySdk.builder()
+ .setPropagators(ContextPropagators.create(W3CTraceContextPropagator.getInstance()))
+ .setTracerProvider(sdkTracerProvider)
+ .build();
+
+ Runtime.getRuntime().addShutdownHook(new Thread(sdkTracerProvider::close));
+ }
+
+ @Override
+ public void shutdown() {
+ //todo: check the spanProcessor if it was already close
+
+ sdkTracerProvider.close();
+
+ //todo: turn the value of useTrace in AbstractHTTPServer into false
+ }
+
+ //Gets or creates a named tracer instance
+ @Override
+ public Tracer getTracer(String instrumentationName) {
+ return openTelemetry.getTracer(instrumentationName);
+ }
+
+ //to inject or extract span context
+ @Override
+ public TextMapPropagator getTextMapPropagator() {
+ return openTelemetry.getPropagators().getTextMapPropagator();
+ }
+}
diff --git a/eventmesh-trace-plugin/eventmesh-trace-zipkin/src/main/java/org/apache/eventmesh/trace/zipkin/config/ZipkinConfiguration.java b/eventmesh-trace-plugin/eventmesh-trace-zipkin/src/main/java/org/apache/eventmesh/trace/zipkin/config/ZipkinConfiguration.java
new file mode 100644
index 0000000..5dfa9a4
--- /dev/null
+++ b/eventmesh-trace-plugin/eventmesh-trace-zipkin/src/main/java/org/apache/eventmesh/trace/zipkin/config/ZipkinConfiguration.java
@@ -0,0 +1,96 @@
+/*
+ * 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.eventmesh.trace.zipkin.config;
+
+import org.apache.commons.lang3.StringUtils;
+
+import java.io.BufferedReader;
+import java.io.File;
+import java.io.FileReader;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.InputStreamReader;
+import java.net.URL;
+import java.util.Properties;
+
+import com.google.common.base.Preconditions;
+
+import lombok.experimental.UtilityClass;
+import lombok.extern.slf4j.Slf4j;
+
+/**
+ * to load the properties form zipkin.properties
+ */
+@Slf4j
+@UtilityClass
+public class ZipkinConfiguration {
+
+ private static final String CONFIG_FILE = "zipkin.properties";
+ private static final Properties properties = new Properties();
+
+ private String eventMeshZipkinIP = "localhost";
+ private int eventMeshZipkinPort = 9411;
+
+ static {
+ loadProperties();
+ initializeConfig();
+ }
+
+ public static String getEventMeshZipkinIP() {
+ return eventMeshZipkinIP;
+ }
+
+ public static int getEventMeshZipkinPort() {
+ return eventMeshZipkinPort;
+ }
+
+ private void initializeConfig() {
+ String eventMeshZipkinIPStr = properties.getProperty("eventmesh.trace.zipkin.ip");
+ Preconditions.checkState(StringUtils.isNotEmpty(eventMeshZipkinIPStr),
+ String.format("%s error", "eventmesh.trace.zipkin.ip"));
+ eventMeshZipkinIP = StringUtils.deleteWhitespace(eventMeshZipkinIPStr);
+
+ String eventMeshZipkinPortStr = properties.getProperty("eventmesh.trace.zipkin.port");
+ if (StringUtils.isNotEmpty(eventMeshZipkinPortStr)) {
+ eventMeshZipkinPort = Integer.parseInt(StringUtils.deleteWhitespace(eventMeshZipkinPortStr));
+ }
+ }
+
+ private void loadProperties() {
+ URL resource = ZipkinConfiguration.class.getClassLoader().getResource(CONFIG_FILE);
+ if (resource != null) {
+ try (InputStream inputStream = resource.openStream()) {
+ if (inputStream.available() > 0) {
+ properties.load(new BufferedReader(new InputStreamReader(inputStream)));
+ }
+ } catch (IOException e) {
+ throw new RuntimeException("Load zipkin.properties file from classpath error");
+ }
+ }
+ // get from config home
+ try {
+ String configPath =
+ System.getProperty("confPath", System.getenv("confPath")) + File.separator + CONFIG_FILE;
+ if (new File(configPath).exists()) {
+ properties.load(new BufferedReader(new FileReader(configPath)));
+ }
+ } catch (IOException e) {
+ throw new IllegalArgumentException("Cannot load zipkin.properties file from conf");
+ }
+ }
+}
diff --git a/eventmesh-common/src/test/resources/configuration.properties b/eventmesh-trace-plugin/eventmesh-trace-zipkin/src/main/resources/META-INF/eventmesh/org.apache.eventmesh.trace.api.TraceService
similarity index 68%
copy from eventmesh-common/src/test/resources/configuration.properties
copy to eventmesh-trace-plugin/eventmesh-trace-zipkin/src/main/resources/META-INF/eventmesh/org.apache.eventmesh.trace.api.TraceService
index d4a7542..a6ccf89 100644
--- a/eventmesh-common/src/test/resources/configuration.properties
+++ b/eventmesh-trace-plugin/eventmesh-trace-zipkin/src/main/resources/META-INF/eventmesh/org.apache.eventmesh.trace.api.TraceService
@@ -1,4 +1,3 @@
-#
# 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.
@@ -13,15 +12,5 @@
# 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.
-#
-eventMesh.server.env=value1
-eventMesh.server.idc=value2
-eventMesh.sysid=3
-eventMesh.server.cluster=value4
-eventMesh.server.name=value5
-eventMesh.server.hostIp=value6
-eventMesh.connector.plugin.type=rocketmq
-eventMesh.security.plugin.type=acl
-eventMesh.registry.plugin.type=namesrv
-eventmesh.trace.export.zipkin.ip=localhost
-eventmesh.trace.exporter.type=Zipkin
\ No newline at end of file
+
+zipkin=org.apache.eventmesh.trace.zipkin.ZipkinTraceService
\ No newline at end of file
diff --git a/eventmesh-common/src/test/resources/configuration.properties b/eventmesh-trace-plugin/eventmesh-trace-zipkin/src/main/resources/zipkin.properties
similarity index 68%
copy from eventmesh-common/src/test/resources/configuration.properties
copy to eventmesh-trace-plugin/eventmesh-trace-zipkin/src/main/resources/zipkin.properties
index d4a7542..323c84c 100644
--- a/eventmesh-common/src/test/resources/configuration.properties
+++ b/eventmesh-trace-plugin/eventmesh-trace-zipkin/src/main/resources/zipkin.properties
@@ -14,14 +14,7 @@
# See the License for the specific language governing permissions and
# limitations under the License.
#
-eventMesh.server.env=value1
-eventMesh.server.idc=value2
-eventMesh.sysid=3
-eventMesh.server.cluster=value4
-eventMesh.server.name=value5
-eventMesh.server.hostIp=value6
-eventMesh.connector.plugin.type=rocketmq
-eventMesh.security.plugin.type=acl
-eventMesh.registry.plugin.type=namesrv
-eventmesh.trace.export.zipkin.ip=localhost
-eventmesh.trace.exporter.type=Zipkin
\ No newline at end of file
+
+#zipkin's working ip and port
+eventmesh.trace.zipkin.ip=localhost
+eventmesh.trace.zipkin.port=9411
\ No newline at end of file
diff --git a/eventmesh-common/src/test/resources/configuration.properties b/eventmesh-trace-plugin/gradle.properties
similarity index 68%
copy from eventmesh-common/src/test/resources/configuration.properties
copy to eventmesh-trace-plugin/gradle.properties
index d4a7542..a9fd83f 100644
--- a/eventmesh-common/src/test/resources/configuration.properties
+++ b/eventmesh-trace-plugin/gradle.properties
@@ -13,15 +13,4 @@
# 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.
-#
-eventMesh.server.env=value1
-eventMesh.server.idc=value2
-eventMesh.sysid=3
-eventMesh.server.cluster=value4
-eventMesh.server.name=value5
-eventMesh.server.hostIp=value6
-eventMesh.connector.plugin.type=rocketmq
-eventMesh.security.plugin.type=acl
-eventMesh.registry.plugin.type=namesrv
-eventmesh.trace.export.zipkin.ip=localhost
-eventmesh.trace.exporter.type=Zipkin
\ No newline at end of file
+#
\ No newline at end of file
diff --git a/settings.gradle b/settings.gradle
index 99694c1..6481fdd 100644
--- a/settings.gradle
+++ b/settings.gradle
@@ -42,4 +42,7 @@ findProject(':eventmesh-protocol-plugin:eventmesh-protocol-grpcmessage')?.name =
include 'eventmesh-metrics-plugin'
include 'eventmesh-metrics-plugin:eventmesh-metrics-api'
include 'eventmesh-metrics-plugin:eventmesh-metrics-opentelemetry'
+include 'eventmesh-trace-plugin'
+include 'eventmesh-trace-plugin:eventmesh-trace-api'
+include 'eventmesh-trace-plugin:eventmesh-trace-zipkin'
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@eventmesh.apache.org
For additional commands, e-mail: commits-help@eventmesh.apache.org