apache · xwm1992 · Dec 16, 2021 · Sep 15, 2021 · Sep 17, 2021 · Sep 20, 2021
diff --git a/build.gradle b/build.gradle
@@ -432,6 +432,8 @@ subprojects {
             dependency 'io.opentelemetry:opentelemetry-exporter-prometheus:1.3.0-alpha'
             dependency 'io.prometheus:simpleclient:0.8.1'
             dependency 'io.prometheus:simpleclient_httpserver:0.8.1'
+            dependency 'io.opentelemetry:opentelemetry-exporter-zipkin:1.3.0'
+            dependency 'io.opentelemetry:opentelemetry-semconv:1.3.0-alpha'
 
             dependency "io.openmessaging:openmessaging-api:2.2.1-pubsub"
 

diff --git a/docs/en/features/eventmesh-trace-design.md b/docs/en/features/eventmesh-trace-design.md
@@ -0,0 +1,89 @@
+# eventmesh-HTTP-trace-design
+
+## Introduction
+
+[EventMesh(incubating)](https://github.com/apache/incubator-eventmesh) is a dynamic cloud-native eventing infrastructure.
+
+## An overview of OpenTelemetry
+
+OpenTelemetry is a collection of tools, APIs, and SDKs. You can use it to instrument, generate, collect, and export telemetry data (metrics, logs, and traces) for analysis in order to understand your software's performance and behavior.
+
+## Requirements
+
+- set tracer
+- different exporter
+- start and end span in server
+
+## Design Details
+
+* SpanProcessor:   BatchSpanProcessor
+
+* Exporter:  log(default), would be changed from properties
+
+```java
+// Configure the batch spans processor. This span processor exports span in batches.
+BatchSpanProcessor batchSpansProcessor =
+    BatchSpanProcessor.builder(exporter)
+        .setMaxExportBatchSize(512) // set the maximum batch size to use
+        .setMaxQueueSize(2048) // set the queue size. This must be >= the export batch size
+        .setExporterTimeout(
+            30, TimeUnit.SECONDS) // set the max amount of time an export can run before getting
+        // interrupted
+        .setScheduleDelay(5, TimeUnit.SECONDS) // set time between two different exports
+        .build();
+OpenTelemetrySdk.builder()
+    .setTracerProvider(
+        SdkTracerProvider.builder().addSpanProcessor(batchSpansProcessor).build())
+    .build();
+```
+
+1. When using the method 'init()' of the class "EventMeshHTTPServer", the class "AbstractHTTPServer” will get the tracer
+
+```java
+super.openTelemetryTraceFactory = new OpenTelemetryTraceFactory(eventMeshHttpConfiguration);
+super.tracer = openTelemetryTraceFactory.getTracer(this.getClass().toString());
+super.textMapPropagator = openTelemetryTraceFactory.getTextMapPropagator();
+```
+
+2. then the trace in class "AbstractHTTPServer” will work.
+
+## Problems
+
+#### How to set different exporter in class 'OpenTelemetryTraceFactory'?(Solved)
+
+After I get the exporter type from properties, how to deal with it.
+
+The 'logExporter' only needs to new it.
+
+But the 'zipkinExporter' needs to new and use the "getZipkinExporter()" method.
+
+## Solutions
+#### Solution of different exporter
+Use reflection to get an exporter.
+
+First of all, different exporter must implement the interface 'EventMeshExporter'.
+
+Then we get the exporter name from the configuration and reflect to the class.
+```java
+//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);
+```
+
+Additional, this will surround with try catch.If the specified exporter cannot be obtained successfully, the default exporter log will be used instead
+
+#### Improvement of different exporter
+
+SPI(To be completed)
+
+
+## Appendix
+
+#### References
+
+https://github.com/open-telemetry/docs-cn/blob/main/QUICKSTART.md
+
+https://github.com/open-telemetry/opentelemetry-java-instrumentation/tree/main/instrumentation/netty
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
@@ -32,6 +32,13 @@ public class CommonConfiguration {
     public String eventMeshSecurityPluginType = "security";
     public int eventMeshPrometheusPort = 19090;
     public String eventMeshRegistryPluginType = "namesrv";
+    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 namesrvAddr = "";
     public Integer eventMeshRegisterIntervalInMills = 10 * 1000;
@@ -48,54 +55,125 @@ public CommonConfiguration(ConfigurationWrapper configurationWrapper) {
     public void init() {
 
         if (configurationWrapper != null) {
-            String eventMeshEnvStr = configurationWrapper.getProp(ConfKeys.KEYS_EVENTMESH_ENV);
-            Preconditions.checkState(StringUtils.isNotEmpty(eventMeshEnvStr), String.format("%s error", ConfKeys.KEYS_EVENTMESH_ENV));
+            String eventMeshEnvStr =
+                    configurationWrapper.getProp(ConfKeys.KEYS_EVENTMESH_ENV);
+            Preconditions.checkState(StringUtils.isNotEmpty(eventMeshEnvStr),
+                    String.format("%s error", ConfKeys.KEYS_EVENTMESH_ENV));
             eventMeshEnv = StringUtils.deleteWhitespace(eventMeshEnvStr);
 
-            String sysIdStr = configurationWrapper.getProp(ConfKeys.KEYS_EVENTMESH_SYSID);
-            Preconditions.checkState(StringUtils.isNotEmpty(sysIdStr) && StringUtils.isNumeric(sysIdStr), String.format("%s error", ConfKeys.KEYS_EVENTMESH_SYSID));
+            String sysIdStr =
+                    configurationWrapper.getProp(ConfKeys.KEYS_EVENTMESH_SYSID);
+            Preconditions.checkState(StringUtils.isNotEmpty(sysIdStr) && StringUtils.isNumeric(sysIdStr),
+                    String.format("%s error", ConfKeys.KEYS_EVENTMESH_SYSID));
             sysID = StringUtils.deleteWhitespace(sysIdStr);
 
-            String eventMeshClusterStr = configurationWrapper.getProp(ConfKeys.KEYS_EVENTMESH_SERVER_CLUSTER);
-            Preconditions.checkState(StringUtils.isNotEmpty(eventMeshClusterStr), String.format("%s error", ConfKeys.KEYS_EVENTMESH_SERVER_CLUSTER));
+            String eventMeshClusterStr =
+                    configurationWrapper.getProp(ConfKeys.KEYS_EVENTMESH_SERVER_CLUSTER);
+            Preconditions.checkState(StringUtils.isNotEmpty(eventMeshClusterStr),
+                    String.format("%s error", ConfKeys.KEYS_EVENTMESH_SERVER_CLUSTER));
             eventMeshCluster = StringUtils.deleteWhitespace(eventMeshClusterStr);
 
-            String eventMeshNameStr = configurationWrapper.getProp(ConfKeys.KEYS_EVENTMESH_SERVER_NAME);
-            Preconditions.checkState(StringUtils.isNotEmpty(eventMeshNameStr), String.format("%s error", ConfKeys.KEYS_EVENTMESH_SERVER_NAME));
+            String eventMeshNameStr =
+                    configurationWrapper.getProp(ConfKeys.KEYS_EVENTMESH_SERVER_NAME);
+            Preconditions.checkState(StringUtils.isNotEmpty(eventMeshNameStr),
+                    String.format("%s error", ConfKeys.KEYS_EVENTMESH_SERVER_NAME));
             eventMeshName = StringUtils.deleteWhitespace(eventMeshNameStr);
 
-            String eventMeshIDCStr = configurationWrapper.getProp(ConfKeys.KEYS_EVENTMESH_IDC);
-            Preconditions.checkState(StringUtils.isNotEmpty(eventMeshIDCStr), String.format("%s error", ConfKeys.KEYS_EVENTMESH_IDC));
-            eventMeshIDC = StringUtils.deleteWhitespace(eventMeshIDCStr);
+            String eventMeshIdcStr =
+                    configurationWrapper.getProp(ConfKeys.KEYS_EVENTMESH_IDC);
+            Preconditions.checkState(StringUtils.isNotEmpty(eventMeshIdcStr),
+                    String.format("%s error", ConfKeys.KEYS_EVENTMESH_IDC));
+            eventMeshIDC = StringUtils.deleteWhitespace(eventMeshIdcStr);
 
-            String eventMeshPrometheusPortStr = configurationWrapper.getProp(ConfKeys.KEY_EVENTMESH_METRICS_PROMETHEUS_PORT);
+            String eventMeshPrometheusPortStr =
+                    configurationWrapper.getProp(ConfKeys.KEY_EVENTMESH_METRICS_PROMETHEUS_PORT);
             if (StringUtils.isNotEmpty(eventMeshPrometheusPortStr)) {
-                eventMeshPrometheusPort = Integer.valueOf(StringUtils.deleteWhitespace(eventMeshPrometheusPortStr));
+                eventMeshPrometheusPort =
+                        Integer.valueOf(StringUtils.deleteWhitespace(eventMeshPrometheusPortStr));
             }
 
-            eventMeshServerIp = configurationWrapper.getProp(ConfKeys.KEYS_EVENTMESH_SERVER_HOST_IP);
+            eventMeshServerIp =
+                    configurationWrapper.getProp(ConfKeys.KEYS_EVENTMESH_SERVER_HOST_IP);
             if (StringUtils.isBlank(eventMeshServerIp)) {
                 eventMeshServerIp = IPUtils.getLocalAddress();
             }
 
-            eventMeshConnectorPluginType = configurationWrapper.getProp(ConfKeys.KEYS_ENENTMESH_CONNECTOR_PLUGIN_TYPE);
-            Preconditions.checkState(StringUtils.isNotEmpty(eventMeshConnectorPluginType), String.format("%s error", ConfKeys.KEYS_ENENTMESH_CONNECTOR_PLUGIN_TYPE));
+            eventMeshConnectorPluginType =
+                    configurationWrapper.getProp(ConfKeys.KEYS_ENENTMESH_CONNECTOR_PLUGIN_TYPE);
+            Preconditions.checkState(StringUtils.isNotEmpty(eventMeshConnectorPluginType),
+                    String.format("%s error", ConfKeys.KEYS_ENENTMESH_CONNECTOR_PLUGIN_TYPE));
 
-            String eventMeshServerAclEnableStr = configurationWrapper.getProp(ConfKeys.KEYS_EVENTMESH_SECURITY_ENABLED);
+            String eventMeshServerAclEnableStr =
+                    configurationWrapper.getProp(ConfKeys.KEYS_EVENTMESH_SECURITY_ENABLED);
             if (StringUtils.isNotBlank(eventMeshServerAclEnableStr)) {
-                eventMeshServerSecurityEnable = Boolean.valueOf(StringUtils.deleteWhitespace(eventMeshServerAclEnableStr));
+                eventMeshServerSecurityEnable =
+                        Boolean.valueOf(StringUtils.deleteWhitespace(eventMeshServerAclEnableStr));
             }
 
-            eventMeshSecurityPluginType = configurationWrapper.getProp(ConfKeys.KEYS_ENENTMESH_SECURITY_PLUGIN_TYPE);
-            Preconditions.checkState(StringUtils.isNotEmpty(eventMeshSecurityPluginType), String.format("%s error", ConfKeys.KEYS_ENENTMESH_SECURITY_PLUGIN_TYPE));
+            eventMeshSecurityPluginType =
+                    configurationWrapper.getProp(ConfKeys.KEYS_ENENTMESH_SECURITY_PLUGIN_TYPE);
+            Preconditions.checkState(StringUtils.isNotEmpty(eventMeshSecurityPluginType),
+                    String.format("%s error", ConfKeys.KEYS_ENENTMESH_SECURITY_PLUGIN_TYPE));
 
-            String eventMeshServerRegistryEnableStr = configurationWrapper.getProp(ConfKeys.KEYS_EVENTMESH_REGISTRY_ENABLED);
+            String eventMeshServerRegistryEnableStr =
+                    configurationWrapper.getProp(ConfKeys.KEYS_EVENTMESH_REGISTRY_ENABLED);
             if (StringUtils.isNotBlank(eventMeshServerRegistryEnableStr)) {
-                eventMeshServerRegistryEnable = Boolean.valueOf(StringUtils.deleteWhitespace(eventMeshServerRegistryEnableStr));
+                eventMeshServerRegistryEnable =
+                        Boolean.valueOf(StringUtils.deleteWhitespace(eventMeshServerRegistryEnableStr));
             }
 
-            eventMeshRegistryPluginType = configurationWrapper.getProp(ConfKeys.KEYS_ENENTMESH_REGISTRY_PLUGIN_TYPE);
-            Preconditions.checkState(StringUtils.isNotEmpty(eventMeshRegistryPluginType), String.format("%s error", ConfKeys.KEYS_ENENTMESH_REGISTRY_PLUGIN_TYPE));
+            eventMeshRegistryPluginType =
+                    configurationWrapper.getProp(ConfKeys.KEYS_ENENTMESH_REGISTRY_PLUGIN_TYPE);
+            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.valueOf(StringUtils.deleteWhitespace(eventMeshTraceExporterMaxExportSizeStr));
+            }
+
+            String eventMeshTraceExporterMaxQueueSizeStr =
+                    configurationWrapper.getProp(ConfKeys.KEYS_EVENTMESH_TRACE_EXPORTER_MAX_QUEUE_SIZE);
+            if (StringUtils.isNotEmpty(eventMeshTraceExporterMaxQueueSizeStr)) {
+                eventMeshTraceExporterMaxQueueSize =
+                        Integer.valueOf(StringUtils.deleteWhitespace(eventMeshTraceExporterMaxQueueSizeStr));
+            }
+
+            String eventMeshTraceExporterExportTimeoutStr =
+                    configurationWrapper.getProp(ConfKeys.KEYS_EVENTMESH_TRACE_EXPORTER_EXPORT_TIMEOUT);
+            if (StringUtils.isNotEmpty(eventMeshTraceExporterExportTimeoutStr)) {
+                eventMeshTraceExporterExportTimeout =
+                        Integer.valueOf(StringUtils.deleteWhitespace(eventMeshTraceExporterExportTimeoutStr));
+            }
+
+            String eventMeshTraceExporterExportIntervalStr =
+                    configurationWrapper.getProp(ConfKeys.KEYS_EVENTMESH_TRACE_EXPORTER_EXPORT_INTERVAL);
+            if (StringUtils.isNotEmpty(eventMeshTraceExporterExportIntervalStr)) {
+                eventMeshTraceExporterExportInterval =
+                        Integer.valueOf(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.valueOf(StringUtils.deleteWhitespace(eventMeshTraceExportZipkinPortStr));
+            }
         }
     }
 
@@ -112,9 +190,11 @@ static class ConfKeys {
 
         public static String KEYS_EVENTMESH_SERVER_HOST_IP = "eventMesh.server.hostIp";
 
-        public static String KEYS_EVENTMESH_SERVER_REGISTER_INTERVAL = "eventMesh.server.registry.registerIntervalInMills";
+        public static String KEYS_EVENTMESH_SERVER_REGISTER_INTERVAL =
+                "eventMesh.server.registry.registerIntervalInMills";
 
-        public static String KEYS_EVENTMESH_SERVER_FETCH_REGISTRY_ADDR_INTERVAL = "eventMesh.server.registry.fetchRegistryAddrIntervalInMills";
+        public static String KEYS_EVENTMESH_SERVER_FETCH_REGISTRY_ADDR_INTERVAL =
+                "eventMesh.server.registry.fetchRegistryAddrIntervalInMills";
 
         public static String KEYS_ENENTMESH_CONNECTOR_PLUGIN_TYPE = "eventMesh.connector.plugin.type";
 
@@ -127,5 +207,19 @@ static class ConfKeys {
         public static String KEYS_EVENTMESH_REGISTRY_ENABLED = "eventMesh.server.registry.enabled";
 
         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";
     }
 }
diff --git a/eventmesh-common/src/test/resources/configuration.properties b/eventmesh-common/src/test/resources/configuration.properties
@@ -23,4 +23,6 @@ eventMesh.server.name=value5
 eventMesh.server.hostIp=value6
 eventMesh.connector.plugin.type=rocketmq
 eventMesh.security.plugin.type=acl
-eventMesh.registry.plugin.type=namesrv
+eventMesh.registry.plugin.type=namesrv
+eventmesh.trace.export.zipkin.ip=localhost
+eventmesh.trace.exporter.type=Zipkin
diff --git a/eventmesh-runtime/build.gradle b/eventmesh-runtime/build.gradle
@@ -23,6 +23,9 @@ dependencies {
     implementation 'io.prometheus:simpleclient'
     implementation 'io.prometheus:simpleclient_httpserver'
     implementation 'io.cloudevents:cloudevents-core'
+    implementation 'io.opentelemetry:opentelemetry-exporter-zipkin'
+    implementation 'io.opentelemetry:opentelemetry-semconv'
+
 
     implementation "org.apache.httpcomponents:httpclient"
     implementation 'io.netty:netty-all'

diff --git a/eventmesh-runtime/conf/eventmesh.properties b/eventmesh-runtime/conf/eventmesh.properties
@@ -72,4 +72,20 @@ eventMesh.server.registry.enabled=false
 eventMesh.registry.plugin.type=namesrv
 
 #prometheusPort
-eventMesh.metrics.prometheus.port=19090
+eventMesh.metrics.prometheus.port=19090
+
+#trace exporter
+eventmesh.trace.exporter.type=Zipkin
+
+#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