bugfix for TelemetryInterceptor

dinstone · Jan 16, 2024 · 6be2a63 · 6be2a63
1 parent bf35275
commit 6be2a63
Show file tree

Hide file tree

Showing 6 changed files with 118 additions and 76 deletions.
diff --git a/.../focus-server-core/src/main/java/com/dinstone/focus/server/invoke/LocalInvokeHandler.java b/.../focus-server-core/src/main/java/com/dinstone/focus/server/invoke/LocalInvokeHandler.java
@@ -20,11 +20,13 @@
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.Future;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
 
 import com.dinstone.focus.config.MethodConfig;
 import com.dinstone.focus.config.ServiceConfig;
 import com.dinstone.focus.exception.BusinessException;
 import com.dinstone.focus.exception.ErrorCode;
+import com.dinstone.focus.exception.InvokeException;
 import com.dinstone.focus.exception.ServiceException;
 import com.dinstone.focus.invoke.Handler;
 import com.dinstone.focus.protocol.Call;
@@ -68,6 +70,8 @@ public CompletableFuture<Reply> handle(Call call) throws Exception {
             throw new ServiceException(ErrorCode.PARAM_ERROR, e);
         } catch (IllegalAccessException e) {
             throw new ServiceException(ErrorCode.ACCESS_ERROR, e);
+        } catch (TimeoutException e) {
+            throw new InvokeException(ErrorCode.TIMEOUT_ERROR, e);
         }
     }
 

diff --git a/focus-telemetry/pom.xml b/focus-telemetry/pom.xml
@@ -1,55 +1,61 @@
 <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
-	<modelVersion>4.0.0</modelVersion>
-	<parent>
-		<groupId>com.dinstone.focus</groupId>
-		<artifactId>focus-parent</artifactId>
-		<version>1.2.0</version>
-	</parent>
-	<artifactId>focus-telemetry</artifactId>
-	<dependencyManagement>
-		<dependencies>
-			<dependency>
-				<groupId>io.opentelemetry</groupId>
-				<artifactId>opentelemetry-bom</artifactId>
-				<version>1.30.0</version>
-				<type>pom</type>
-				<scope>import</scope>
-			</dependency>
-		</dependencies>
-	</dependencyManagement>
-	<dependencies>
-		<dependency>
-			<groupId>com.dinstone.focus</groupId>
-			<artifactId>focus-core</artifactId>
-			<version>${project.version}</version>
-		</dependency>
-		<dependency>
-			<groupId>io.opentelemetry</groupId>
-			<artifactId>opentelemetry-api</artifactId>
-		</dependency>
-		<dependency>
-			<groupId>io.opentelemetry</groupId>
-			<artifactId>opentelemetry-sdk</artifactId>
-		</dependency>
-		<dependency>
-			<groupId>io.opentelemetry</groupId>
-			<artifactId>opentelemetry-exporter-logging</artifactId>
-			<scope>test</scope>
-		</dependency>
-		<dependency>
-			<groupId>io.opentelemetry</groupId>
-			<artifactId>opentelemetry-exporter-zipkin</artifactId>
-		</dependency>
-	</dependencies>
-	<build>
-		<plugins>
-			<plugin>
-				<groupId>net.revelc.code.formatter</groupId>
-				<artifactId>formatter-maven-plugin</artifactId>
-				<configuration>
-					<configFile>../guide/formatter-java.xml</configFile>
-				</configuration>
-			</plugin>
-		</plugins>
-	</build>
+    <modelVersion>4.0.0</modelVersion>
+    <parent>
+        <groupId>com.dinstone.focus</groupId>
+        <artifactId>focus-parent</artifactId>
+        <version>1.2.0</version>
+    </parent>
+    <artifactId>focus-telemetry</artifactId>
+    <dependencyManagement>
+        <dependencies>
+            <dependency>
+                <groupId>io.opentelemetry</groupId>
+                <artifactId>opentelemetry-bom</artifactId>
+                <version>1.33.0</version>
+                <type>pom</type>
+                <scope>import</scope>
+            </dependency>
+        </dependencies>
+    </dependencyManagement>
+    <dependencies>
+        <dependency>
+            <groupId>com.dinstone.focus</groupId>
+            <artifactId>focus-core</artifactId>
+            <version>${project.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>io.opentelemetry</groupId>
+            <artifactId>opentelemetry-api</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>io.opentelemetry</groupId>
+            <artifactId>opentelemetry-sdk</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>io.opentelemetry</groupId>
+            <artifactId>opentelemetry-exporter-logging</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>io.opentelemetry</groupId>
+            <artifactId>opentelemetry-exporter-zipkin</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>io.opentelemetry</groupId>
+            <artifactId>opentelemetry-exporter-jaeger</artifactId>
+            <scope>test</scope>
+        </dependency>
+    </dependencies>
+    <build>
+        <plugins>
+            <plugin>
+                <groupId>net.revelc.code.formatter</groupId>
+                <artifactId>formatter-maven-plugin</artifactId>
+                <configuration>
+                    <configFile>../guide/formatter-java.xml</configFile>
+                </configuration>
+            </plugin>
+        </plugins>
+    </build>
 </project>
diff --git a/focus-telemetry/src/main/java/com/dinstone/focus/telemetry/TelemetryInterceptor.java b/focus-telemetry/src/main/java/com/dinstone/focus/telemetry/TelemetryInterceptor.java
@@ -17,6 +17,7 @@
 
 import java.util.concurrent.CompletableFuture;
 
+import com.dinstone.focus.exception.InvokeException;
 import com.dinstone.focus.invoke.Handler;
 import com.dinstone.focus.invoke.Interceptor;
 import com.dinstone.focus.protocol.Call;
@@ -75,15 +76,18 @@ public void set(Call carrier, String key, String value) {
     public CompletableFuture<Reply> intercept(Call call, Handler chain) throws Exception {
         Tracer tracer = telemetry.getTracer(call.getService());
         if (kind == Kind.SERVER) {
-            // Extract the SpanContext and other elements from the request.
-            Context ec = telemetry.getPropagators().getTextMapPropagator().extract(Context.current(), call, getter);
-            try (Scope ss = ec.makeCurrent()) {
-                return chain.handle(call);
+            Span span = getServerSpan(call, tracer);
+            try (Scope ignored = span.makeCurrent()) {
+                return chain.handle(call).whenComplete((reply, error) -> {
+                    finishSpan(reply, error, span);
+                });
+            } catch (Throwable error) {
+                finishSpan(null, error, span);
+                throw error;
             }
         } else {
-            Span span = tracer.spanBuilder(call.getMethod()).setSpanKind(SpanKind.CLIENT).startSpan();
-            try (Scope ss = span.makeCurrent()) {
-                span.setAttribute(RPC_SERVICE, call.getService()).setAttribute(RPC_METHOD, call.getMethod());
+            Span span = getClientSpan(call, tracer);
+            try (Scope ignored = span.makeCurrent()) {
                 // Inject the request with the *current* Context, which contains our current
                 // Span.
                 telemetry.getPropagators().getTextMapPropagator().inject(Context.current(), call, setter);
@@ -98,10 +102,26 @@ public CompletableFuture<Reply> intercept(Call call, Handler chain) throws Excep
 
     }
 
+    private Span getClientSpan(Call call, Tracer tracer) {
+        Span span = tracer.spanBuilder(call.getEndpoint()).setSpanKind(SpanKind.CLIENT).startSpan();
+        return span.setAttribute(RPC_SERVICE, call.getService()).setAttribute(RPC_METHOD, call.getMethod());
+    }
+
+    private Span getServerSpan(Call call, Tracer tracer) {
+        // Extract the SpanContext and other elements from the request.
+        Context pc = telemetry.getPropagators().getTextMapPropagator().extract(Context.current(), call, getter);
+        Span span = tracer.spanBuilder(call.getEndpoint()).setSpanKind(SpanKind.SERVER).setParent(pc).startSpan();
+        return span.setAttribute(RPC_SERVICE, call.getService()).setAttribute(RPC_METHOD, call.getMethod());
+    }
+
     private void finishSpan(Reply reply, Throwable error, Span span) {
         if (error != null) {
             span.setStatus(StatusCode.ERROR, error.getMessage());
             span.recordException(error);
+        } else if (reply != null && reply.isError()) {
+            final InvokeException data = (InvokeException) reply.getData();
+            span.setStatus(StatusCode.ERROR, data.getMessage());
+            span.recordException(data);
         }
         span.end();
     }

diff --git a/focus-telemetry/src/test/java/focus/TelemetryTest.java b/focus-telemetry/src/test/java/focus/TelemetryTest.java
@@ -15,16 +15,24 @@
  */
 package focus;
 
+import java.io.IOException;
+import java.time.Duration;
+
 import io.opentelemetry.api.OpenTelemetry;
 import io.opentelemetry.api.common.AttributeKey;
 import io.opentelemetry.api.common.Attributes;
+import io.opentelemetry.api.metrics.LongHistogram;
+import io.opentelemetry.api.metrics.Meter;
 import io.opentelemetry.api.trace.Span;
 import io.opentelemetry.api.trace.Tracer;
 import io.opentelemetry.api.trace.propagation.W3CTraceContextPropagator;
 import io.opentelemetry.context.Scope;
 import io.opentelemetry.context.propagation.ContextPropagators;
+import io.opentelemetry.exporter.logging.LoggingMetricExporter;
 import io.opentelemetry.exporter.logging.LoggingSpanExporter;
 import io.opentelemetry.sdk.OpenTelemetrySdk;
+import io.opentelemetry.sdk.metrics.SdkMeterProvider;
+import io.opentelemetry.sdk.metrics.export.PeriodicMetricReader;
 import io.opentelemetry.sdk.resources.Resource;
 import io.opentelemetry.sdk.trace.SdkTracerProvider;
 import io.opentelemetry.sdk.trace.export.BatchSpanProcessor;
@@ -33,10 +41,16 @@ public class TelemetryTest {
 
     private static Tracer tracer;
 
-    public static void main(String[] args) {
+    public static void main(String[] args) throws IOException {
         OpenTelemetry openTelemetry = getOpenTelemetry();
 
-        tracer = openTelemetry.getTracer("telemetry-test-service", "1.0.0");
+        String serviceName = "telemetry-test-service";
+        Meter meter = openTelemetry.getMeter(serviceName);
+        tracer = openTelemetry.getTracer(serviceName, "1.0.0");
+
+        LongHistogram h = meter.histogramBuilder("main").ofLongs().build();
+
+        h.record(2000);
 
         Span span = tracer.spanBuilder("main").startSpan();
         // Make the span the current span
@@ -49,6 +63,8 @@ public static void main(String[] args) {
             span.end();
         }
 
+        System.in.read();
+
         System.out.println("over");
     }
 
@@ -81,10 +97,13 @@ private static OpenTelemetry getOpenTelemetry() {
                 .addSpanProcessor(BatchSpanProcessor.builder(LoggingSpanExporter.create()).build())
                 .setResource(resource).build();
 
-        OpenTelemetry openTelemetry = OpenTelemetrySdk.builder().setTracerProvider(sdkTracerProvider)
+        SdkMeterProvider sdkMeterProvider = SdkMeterProvider.builder().registerMetricReader(PeriodicMetricReader
+                .builder(LoggingMetricExporter.create()).setInterval(Duration.ofMillis(3000000)).build())
+                .setResource(resource).build();
+
+        return OpenTelemetrySdk.builder().setTracerProvider(sdkTracerProvider).setMeterProvider(sdkMeterProvider)
                 .setPropagators(ContextPropagators.create(W3CTraceContextPropagator.getInstance()))
                 .buildAndRegisterGlobal();
-        return openTelemetry;
     }
 
 }
diff --git a/focus-transport/focus-transport-photon/pom.xml b/focus-transport/focus-transport-photon/pom.xml
@@ -15,7 +15,7 @@
 		<dependency>
 			<groupId>com.dinstone.photon</groupId>
 			<artifactId>photon</artifactId>
-			<version>1.1.4</version>
+			<version>1.1.5</version>
 		</dependency>
 	</dependencies>
 	<build>

diff --git a/...port-photon/src/main/java/com/dinstone/focus/transport/photon/PhotonMessageProcessor.java b/...port-photon/src/main/java/com/dinstone/focus/transport/photon/PhotonMessageProcessor.java
@@ -15,7 +15,6 @@
  */
 package com.dinstone.focus.transport.photon;
 
-import java.io.ByteArrayOutputStream;
 import java.io.IOException;
 import java.util.concurrent.Executor;
 import java.util.function.Function;
@@ -36,7 +35,6 @@
 import com.dinstone.photon.message.Request;
 import com.dinstone.photon.message.Response;
 import com.dinstone.photon.message.Response.Status;
-import com.dinstone.photon.utils.ByteStreamUtil;
 
 import io.netty.util.CharsetUtil;
 
@@ -125,15 +123,10 @@ private Response encode(Reply reply, ServiceConfig serviceConfig, MethodConfig m
         Response response = new Response();
         if (reply.isError()) {
             response.setStatus(Status.FAILURE);
-            try {
-                InvokeException exception = (InvokeException) reply.getData();
-                ByteArrayOutputStream bao = new ByteArrayOutputStream();
-                ByteStreamUtil.writeInt(bao, exception.getCode().value());
-                ByteStreamUtil.writeString(bao, exception.getMessage());
-                response.setContent(bao.toByteArray());
-            } catch (IOException e) {
-                throw new ServiceException(ErrorCode.CODEC_ERROR,
-                        "serialize encode error: " + methodConfig.getMethodName(), e);
+            InvokeException exception = (InvokeException) reply.getData();
+            response.headers().setInt(InvokeException.CODE_KEY, exception.getCode().value());
+            if (exception.getMessage() != null) {
+                response.setContent(exception.getMessage().getBytes(CharsetUtil.UTF_8));
             }
         } else {
             response.setStatus(Status.SUCCESS);