[ISSUE #2789] Add Jaeger tracing plugin service

build.gradle

@@ -465,7 +465,7 @@ subprojects {
             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.opentelemetry:opentelemetry-exporter-jaeger:1.20.1'
+            dependency 'io.opentelemetry:opentelemetry-exporter-jaeger:1.4.0'
 
             dependency "io.openmessaging:openmessaging-api:2.2.1-pubsub"
 

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

@@ -25,6 +25,12 @@ dependencies {
     implementation 'io.opentelemetry:opentelemetry-exporter-jaeger'
     implementation 'io.opentelemetry:opentelemetry-semconv'
 
+    // grpc
+    implementation "io.grpc:grpc-protobuf"
+    implementation "io.grpc:grpc-stub"
+    implementation "io.grpc:grpc-netty"
+    implementation "io.grpc:grpc-netty-shaded"
+
     compileOnly 'org.projectlombok:lombok:1.18.22'
     annotationProcessor 'org.projectlombok:lombok:1.18.22'
 

eventmesh-trace-plugin/eventmesh-trace-jaeger/src/main/java/org/apache/eventmesh/trace/jaeger/JaegerTraceService.java

@@ -17,46 +17,152 @@
 
 package org.apache.eventmesh.trace.jaeger;
 
+import static io.opentelemetry.api.common.AttributeKey.stringKey;
+
 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.jaeger.common.JaegerConstants;
+import org.apache.eventmesh.trace.jaeger.config.JaegerConfiguration;
 
 import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
 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.exporter.jaeger.JaegerGrpcSpanExporter;
+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 JaegerTraceService implements EventMeshTraceService {
 
+    private String eventMeshJaegerIp;
+
+    private int eventMeshJaegerPort;
+
+    private int eventMeshTraceExportInterval;
+
+    private int eventMeshTraceExportTimeout;
+
+    private int eventMeshTraceMaxExportSize;
+
+    private int eventMeshTraceMaxQueueSize;
+
+    protected SdkTracerProvider sdkTracerProvider;
+
+    protected OpenTelemetry openTelemetry;
+
+    protected Thread shutdownHook;
+
+    private Tracer tracer;
+
+    private TextMapPropagator textMapPropagator;
+
     @Override
     public void init() throws TraceException {
+        // jaeger's config
+        eventMeshJaegerIp = JaegerConfiguration.getEventMeshJaegerIp();
+        eventMeshJaegerPort = JaegerConfiguration.getEventMeshJaegerPort();
+        // exporter's config
+        eventMeshTraceExportInterval = ExporterConfiguration.getEventMeshTraceExportInterval();
+        eventMeshTraceExportTimeout = ExporterConfiguration.getEventMeshTraceExportTimeout();
+        eventMeshTraceMaxExportSize = ExporterConfiguration.getEventMeshTraceMaxExportSize();
+        eventMeshTraceMaxQueueSize = ExporterConfiguration.getEventMeshTraceMaxQueueSize();
+
+        String httpEndpoint = String.format("http://%s:%s", eventMeshJaegerIp, eventMeshJaegerPort);
+        JaegerGrpcSpanExporter jaegerExporter = JaegerGrpcSpanExporter.builder()
+            .setEndpoint(httpEndpoint)
+            .build();
+
+        SpanProcessor spanProcessor = BatchSpanProcessor.builder(jaegerExporter)
+            .setScheduleDelay(eventMeshTraceExportInterval, TimeUnit.SECONDS)
+            .setExporterTimeout(eventMeshTraceExportTimeout, TimeUnit.SECONDS)
+            .setMaxExportBatchSize(eventMeshTraceMaxExportSize)
+            .setMaxQueueSize(eventMeshTraceMaxQueueSize)
+            .build();
+
+        Resource serviceNameResource =
+            Resource.create(Attributes.of(stringKey("service.name"), JaegerConstants.SERVICE_NAME));
+
+        sdkTracerProvider = SdkTracerProvider.builder()
+            .addSpanProcessor(spanProcessor)
+            .setResource(Resource.getDefault().merge(serviceNameResource))
+            .build();
 
+        openTelemetry = OpenTelemetrySdk.builder()
+            .setPropagators(ContextPropagators.create(W3CTraceContextPropagator.getInstance()))
+            .setTracerProvider(sdkTracerProvider)
+            .build();
+
+        tracer = openTelemetry.getTracer(JaegerConstants.SERVICE_NAME);
+        textMapPropagator = openTelemetry.getPropagators().getTextMapPropagator();
+
+        shutdownHook = new Thread(sdkTracerProvider::close);
+        Runtime.getRuntime().addShutdownHook(shutdownHook);
     }
 
     @Override
     public Context extractFrom(Context context, Map<String, Object> carrier) throws TraceException {
-        return null;
+        textMapPropagator.extract(context, carrier, new TextMapGetter<Map<String, Object>>() {
+            @Override
+            public Iterable<String> keys(Map<String, Object> carrier) {
+                return carrier.keySet();
+            }
+
+            @Nullable
+            @Override
+            public String get(@Nullable Map<String, Object> carrier, String key) {
+                return Optional.ofNullable(carrier.get(key)).map(Objects::toString).orElse(null);
+            }
+        });
+        return context;
     }
 
     @Override
     public void inject(Context context, Map<String, Object> carrier) {
-
+        textMapPropagator.inject(context, carrier, (cr, key, value) -> {
+            if (cr != null) {
+                cr.put(key, value);
+            }
+        });
     }
 
     @Override
     public Span createSpan(String spanName, SpanKind spanKind, long startTimestamp, TimeUnit timeUnit, Context context,
                            boolean isSpanFinishInOtherThread) throws TraceException {
-        return null;
+        return tracer.spanBuilder(spanName)
+            .setParent(context)
+            .setSpanKind(spanKind)
+            .setStartTimestamp(startTimestamp, timeUnit)
+            .startSpan();
     }
 
     @Override
     public Span createSpan(String spanName, SpanKind spanKind, Context context, boolean isSpanFinishInOtherThread) throws TraceException {
-        return null;
+        return tracer.spanBuilder(spanName)
+            .setParent(context)
+            .setSpanKind(spanKind)
+            .setStartTimestamp(System.currentTimeMillis(), TimeUnit.MILLISECONDS)
+            .startSpan();
     }
 
     @Override
     public void shutdown() throws TraceException {
-
+        sdkTracerProvider.close();
     }
 }

eventmesh-trace-plugin/eventmesh-trace-jaeger/src/main/java/org/apache/eventmesh/trace/jaeger/config/JaegerConfiguration.java

@@ -63,7 +63,7 @@ private void loadProperties() {
                     PROPERTIES.load(reader);
                 }
             } catch (IOException e) {
-                throw new RuntimeException("Load zipkin.properties file from classpath error", e);
+                throw new RuntimeException("Load jaeger.properties file from classpath error", e);
             }
         }
         // get from config home

eventmesh-trace-plugin/eventmesh-trace-jaeger/src/main/resources/META-INF/eventmesh/org.apache.eventmesh.trace.api.EventMeshTraceService

@@ -0,0 +1,16 @@
+# 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.
+
+jaeger=org.apache.eventmesh.trace.jaeger.JaegerTraceService

eventmesh-trace-plugin/eventmesh-trace-jaeger/src/main/resources/jaeger.properties

@@ -0,0 +1,19 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+eventmesh.trace.jaeger.ip=localhost
+eventmesh.trace.jaeger.port=14250

eventmesh-trace-plugin/eventmesh-trace-jaeger/src/test/java/org/apache/eventmesh/trace/jaeger/JaegerTraceServiceTest.java

@@ -0,0 +1,42 @@
+package org.apache.eventmesh.trace.jaeger;
+
+import static org.junit.Assert.assertThrows;
+
+import java.lang.reflect.Field;
+
+import org.junit.Assert;
+import org.junit.Test;
+import org.mockito.Mockito;
+
+import io.opentelemetry.sdk.trace.SdkTracerProvider;
+
+public class JaegerTraceServiceTest {
+
+    @Test
+    public void testInit() {
+        JaegerTraceService jaegerTraceService = new JaegerTraceService();
+        jaegerTraceService.init();
+
+        Assert.assertNotNull(jaegerTraceService.sdkTracerProvider);
+        Assert.assertNotNull(jaegerTraceService.openTelemetry);
+        Assert.assertNotNull(jaegerTraceService.shutdownHook);
+
+        IllegalArgumentException illegalArgumentException =
+            assertThrows(IllegalArgumentException.class, () ->
+                Runtime.getRuntime().addShutdownHook(jaegerTraceService.shutdownHook));
+        Assert.assertEquals(illegalArgumentException.getMessage(), "Hook previously registered");
+    }
+
+    @Test
+    public void testShutdown() throws NoSuchFieldException, IllegalAccessException {
+        SdkTracerProvider mockSdkTracerProvider = Mockito.mock(SdkTracerProvider.class);
+        JaegerTraceService jaegerTraceService = new JaegerTraceService();
+        jaegerTraceService.init();
+        Field sdkTracerProviderField = JaegerTraceService.class.getDeclaredField("sdkTracerProvider");
+        sdkTracerProviderField.setAccessible(true);
+        sdkTracerProviderField.set(jaegerTraceService, mockSdkTracerProvider);
+
+        jaegerTraceService.shutdown();
+        Mockito.verify(mockSdkTracerProvider, Mockito.times(1)).close();
+    }
+}

eventmesh-trace-plugin/eventmesh-trace-jaeger/src/test/resources/jaeger.properties

@@ -16,4 +16,4 @@
 #
 
 eventmesh.trace.jaeger.ip=localhost
-eventmesh.trace.pinpoint.port=14250
+eventmesh.trace.jaeger.port=14250