Merge pull request #952 from lrhkobe/trace_improve_1

[ISSUE #946] Optimize trace module in eventmesh

eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/EventMeshHTTPServer.java

@@ -53,8 +53,6 @@
 import org.apache.eventmesh.runtime.core.protocol.http.retry.HttpRetryer;
 import org.apache.eventmesh.runtime.metrics.http.HTTPMetricsServer;
 import org.apache.eventmesh.runtime.registry.Registry;
-import org.apache.eventmesh.trace.api.TracePluginFactory;
-import org.apache.eventmesh.trace.api.TraceService;
 
 import org.apache.commons.lang3.StringUtils;
 
@@ -78,8 +76,6 @@ public class EventMeshHTTPServer extends AbstractHTTPServer {
 
     private EventMeshHTTPConfiguration eventMeshHttpConfiguration;
 
-    private TraceService traceService;
-
     private Registry registry;
 
     public final ConcurrentHashMap<String /**group*/, ConsumerGroupConf> localConsumerGroupMapping =
@@ -253,11 +249,6 @@ public void init() throws Exception {
         //get the trace-plugin
         if (StringUtils.isNotEmpty(eventMeshHttpConfiguration.eventMeshTracePluginType) && eventMeshHttpConfiguration.eventMeshServerTraceEnable) {
 
-            traceService =
-                TracePluginFactory.getTraceService(eventMeshHttpConfiguration.eventMeshTracePluginType);
-            traceService.init();
-            super.tracer = traceService.getTracer(super.getClass().toString());
-            super.textMapPropagator = traceService.getTextMapPropagator();
             super.useTrace = true;
         }
 
@@ -284,10 +275,6 @@ public void shutdown() throws Exception {
 
         metrics.shutdown();
 
-        if (traceService != null) {
-            traceService.shutdown();
-        }
-
         consumerManager.shutdown();
 
         shutdownThreadPool();

eventmesh-trace-plugin/eventmesh-trace-api/build.gradle

@@ -18,6 +18,7 @@
 dependencies {
     api project(":eventmesh-spi")
     implementation project(":eventmesh-common")
+    api 'io.cloudevents:cloudevents-core'
 
     implementation 'io.opentelemetry:opentelemetry-api'
     implementation 'io.opentelemetry:opentelemetry-sdk'

eventmesh-trace-plugin/eventmesh-trace-api/src/main/java/org/apache/eventmesh/trace/api/TraceService.java → eventmesh-trace-plugin/eventmesh-trace-api/src/main/java/org/apache/eventmesh/trace/api/EventMeshTraceService.java

@@ -19,37 +19,33 @@
 
 import org.apache.eventmesh.spi.EventMeshExtensionType;
 import org.apache.eventmesh.spi.EventMeshSPI;
+import org.apache.eventmesh.trace.api.exception.TraceException;
 
-import io.opentelemetry.api.trace.Tracer;
-import io.opentelemetry.context.propagation.TextMapPropagator;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+
+import io.opentelemetry.api.trace.Span;
+import io.opentelemetry.api.trace.SpanKind;
+import io.opentelemetry.context.Context;
 
 /**
- * The top-level interface of trace
+ * EventMeshTraceService
  */
 @EventMeshSPI(isSingleton = true, eventMeshExtensionType = EventMeshExtensionType.TRACE)
-public interface TraceService {
-    /**
-     * init the trace service
-     */
-    void init();
-
-    /**
-     * close the trace service
-     */
-    void shutdown();
-
-    /**
-     * get the tracer
-     *
-     * @param instrumentationName
-     * @return
-     */
-    Tracer getTracer(String instrumentationName);
-
-    /**
-     * get TextMapPropagator
-     *
-     * @return
-     */
-    TextMapPropagator getTextMapPropagator();
+public interface EventMeshTraceService {
+    void init() throws TraceException;
+
+    //extract attr from carrier to context
+    Context extractFrom(Context context, Map<String, Object> carrier) throws TraceException;
+
+    //inject attr from context to carrier
+    void inject(Context context, Map<String, Object> carrier);
+
+    Span createSpan(String spanName, SpanKind spanKind, long startTimestamp, TimeUnit timeUnit,
+                    Context context, boolean isSpanFinishInOtherThread) throws TraceException;
+
+    Span createSpan(String spanName, SpanKind spanKind, Context context,
+                    boolean isSpanFinishInOtherThread) throws TraceException;
+
+    void shutdown() throws TraceException;
 }

eventmesh-trace-plugin/eventmesh-trace-api/src/main/java/org/apache/eventmesh/trace/api/TracePluginFactory.java

@@ -34,10 +34,10 @@ public class TracePluginFactory {
      * @param traceServiceType
      * @return
      */
-    public static TraceService getTraceService(String traceServiceType) {
+    public static EventMeshTraceService getEventMeshTraceService(String traceServiceType) {
         checkNotNull(traceServiceType, "traceServiceType cannot be null");
 
-        TraceService traceService = EventMeshExtensionFactory.getExtension(TraceService.class, traceServiceType);
-        return checkNotNull(traceService, "traceServiceType: " + traceServiceType + " is not supported");
+        EventMeshTraceService eventMeshTraceService = EventMeshExtensionFactory.getExtension(EventMeshTraceService.class, traceServiceType);
+        return checkNotNull(eventMeshTraceService, "traceServiceType: " + traceServiceType + " is not supported");
     }
 }

eventmesh-trace-plugin/eventmesh-trace-api/src/main/java/org/apache/eventmesh/trace/api/common/EventMeshTraceConstants.java

@@ -0,0 +1,33 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.eventmesh.trace.api.common;
+
+
+public class EventMeshTraceConstants {
+
+    public static final String TRACE_EVENTMESH_SDK_CLIENT_SPAN = "eventmesh-sdk-client-span";
+
+    public static final String TRACE_UPSTREAM_EVENTMESH_SERVER_SPAN = "upstream-eventmesh-server-span";
+    public static final String TRACE_UPSTREAM_EVENTMESH_CLIENT_SPAN = "upstream-eventmesh-client-span";
+
+    public static final String TRACE_DOWNSTREAM_EVENTMESH_SERVER_SPAN = "downstream-eventmesh-server-span";
+    public static final String TRACE_DOWNSTREAM_EVENTMESH_CLIENT_SPAN = "downstream-eventmesh-client-span";
+
+    public static final String TRACE_EVENTMESH_SDK_SERVER_SPAN = "eventmesh-sdk-server-span";
+
+}

eventmesh-trace-plugin/eventmesh-trace-api/src/main/java/org/apache/eventmesh/trace/api/common/ProtocolType.java

@@ -0,0 +1,23 @@
+/*
+ * 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.common;
+
+public enum ProtocolType {
+    TCP,
+    HTTP
+}

eventmesh-trace-plugin/eventmesh-trace-api/src/main/java/org/apache/eventmesh/trace/api/exception/TraceException.java

@@ -0,0 +1,29 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.eventmesh.trace.api.exception;
+
+public class TraceException extends RuntimeException {
+
+    public TraceException(String message) {
+        super(message);
+    }
+
+    public TraceException(String message, Throwable cause) {
+        super(message, cause);
+    }
+}

eventmesh-trace-plugin/eventmesh-trace-api/src/test/java/org/apache/eventmesh/trace/api/TracePluginFactoryTest.java

@@ -29,18 +29,19 @@ public class TracePluginFactoryTest {
 
     @Test
     public void testFailedGetTraceService() {
-        NullPointerException nullPointerException1 = Assert.assertThrows(NullPointerException.class, () -> TracePluginFactory.getTraceService(null));
+        NullPointerException nullPointerException1 = Assert.assertThrows(NullPointerException.class,
+            () -> TracePluginFactory.getEventMeshTraceService(null));
         MatcherAssert.assertThat(nullPointerException1.getMessage(), is("traceServiceType cannot be null"));
 
         String traceServiceType = "non-Existing";
         NullPointerException nullPointerException2 =
-            Assert.assertThrows(NullPointerException.class, () -> TracePluginFactory.getTraceService(traceServiceType));
+            Assert.assertThrows(NullPointerException.class, () -> TracePluginFactory.getEventMeshTraceService(traceServiceType));
         MatcherAssert.assertThat(nullPointerException2.getMessage(), is("traceServiceType: " + traceServiceType + " is not supported"));
     }
 
     @Test
     public void testSuccessfulGetTraceService() {
-        TraceService zipkinTraceService = TracePluginFactory.getTraceService("zipkin");
+        EventMeshTraceService zipkinTraceService = TracePluginFactory.getEventMeshTraceService("zipkin");
         Assert.assertNotNull(zipkinTraceService);
         Assert.assertTrue(zipkinTraceService instanceof ZipkinTraceService);
     }

eventmesh-trace-plugin/eventmesh-trace-zipkin/src/main/java/org/apache/eventmesh/trace/zipkin/ZipkinTraceService.java

@@ -19,29 +19,39 @@
 
 import static io.opentelemetry.api.common.AttributeKey.stringKey;
 
-import org.apache.eventmesh.trace.api.TraceService;
+import org.apache.eventmesh.trace.api.EventMeshTraceService;
 import org.apache.eventmesh.trace.api.config.ExporterConfiguration;
+import org.apache.eventmesh.trace.api.exception.TraceException;
 import org.apache.eventmesh.trace.zipkin.config.ZipkinConfiguration;
 
+import java.util.Map;
 import java.util.concurrent.TimeUnit;
 
+import javax.annotation.Nullable;
+
 import io.opentelemetry.api.OpenTelemetry;
 import io.opentelemetry.api.common.Attributes;
+import io.opentelemetry.api.trace.Span;
+import io.opentelemetry.api.trace.SpanKind;
 import io.opentelemetry.api.trace.Tracer;
 import io.opentelemetry.api.trace.propagation.W3CTraceContextPropagator;
+import io.opentelemetry.context.Context;
 import io.opentelemetry.context.propagation.ContextPropagators;
+import io.opentelemetry.context.propagation.TextMapGetter;
 import io.opentelemetry.context.propagation.TextMapPropagator;
+import io.opentelemetry.context.propagation.TextMapSetter;
 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 {
+public class ZipkinTraceService implements EventMeshTraceService {
     // Zipkin API Endpoints for uploading spans
     private static final String ENDPOINT_V2_SPANS = "/api/v2/spans";
     // Name of the service(using the instrumentationName)
@@ -58,6 +68,9 @@ public class ZipkinTraceService implements TraceService {
 
     protected Thread shutdownHook;
 
+    private Tracer tracer;
+    private TextMapPropagator textMapPropagator;
+
     @Override
     public void init() {
         //zipkin's config
@@ -94,28 +107,67 @@ public void init() {
             .setTracerProvider(sdkTracerProvider)
             .build();
 
+        //TODO serviceName???
+        tracer = openTelemetry.getTracer(serviceName);
+        textMapPropagator = openTelemetry.getPropagators().getTextMapPropagator();
+
         shutdownHook = new Thread(sdkTracerProvider::close);
         Runtime.getRuntime().addShutdownHook(shutdownHook);
     }
 
     @Override
-    public void shutdown() {
-        //todo: check the spanProcessor if it was already close
+    public Context extractFrom(Context context, Map<String, Object> map) throws TraceException {
+        textMapPropagator.extract(context, map, new TextMapGetter<Map<String, Object>>() {
+            @Override
+            public Iterable<String> keys(Map<String, Object> carrier) {
+                return carrier.keySet();
+            }
+
+            @Override
+            public String get(Map<String, Object> carrier, String key) {
+                return carrier.get(key).toString();
+            }
+        });
+        return context;
+    }
 
-        sdkTracerProvider.close();
+    @Override
+    public void inject(Context context, Map<String, Object> map) {
+        textMapPropagator.inject(context, map, new TextMapSetter<Map<String, Object>>() {
+            @Override
+            public void set(@Nullable Map<String, Object> carrier, String key, String value) {
+                map.put(key, value);
+            }
+        });
+    }
 
-        //todo: turn the value of useTrace in AbstractHTTPServer into false
+    @Override
+    public Span createSpan(String spanName, SpanKind spanKind, long startTime, TimeUnit timeUnit,
+                           Context context, boolean isSpanFinishInOtherThread)
+        throws TraceException {
+        return tracer.spanBuilder(spanName)
+            .setParent(context)
+            .setSpanKind(spanKind)
+            .setStartTimestamp(startTime, timeUnit)
+            .startSpan();
     }
 
-    //Gets or creates a named tracer instance
     @Override
-    public Tracer getTracer(String instrumentationName) {
-        return openTelemetry.getTracer(instrumentationName);
+    public Span createSpan(String spanName, SpanKind spanKind, Context context,
+                           boolean isSpanFinishInOtherThread) throws TraceException {
+        return tracer.spanBuilder(spanName)
+            .setParent(context)
+            .setSpanKind(spanKind)
+            .setStartTimestamp(System.currentTimeMillis(), TimeUnit.MILLISECONDS)
+            .startSpan();
     }
 
-    //to inject or extract span context
     @Override
-    public TextMapPropagator getTextMapPropagator() {
-        return openTelemetry.getPropagators().getTextMapPropagator();
+    public void shutdown() {
+        //todo: check the spanProcessor if it was already close
+
+        sdkTracerProvider.close();
+
+        //todo: turn the value of useTrace in AbstractHTTPServer into false
     }
 }