Skip to content

S
SkyWalking

killuaz丶x / SkyWalking
与 Fork 源项目一致

Fork自 apache / SkyWalking

通知 1
Star 0
Fork 0
S
SkyWalking

体验新版 GitCode，发现更多精彩内容 >>

未验证 提交 97950807 编写于 作者: K Kanro 提交者： GitHub
浏览文件
操作

Enhance gRPC plugin (#4177) 

* Enhance gRPC plugin
Co-authored-by: NKanro <higan@live.cn>
Co-authored-by: Nkezhenxu94 <kezhenxu94@163.com>
Co-authored-by: NGuoDuanLZ <739537111@qq.com>
Co-authored-by: wu-sheng's avatar吴晟 Wu Sheng <wu.sheng@foxmail.com>
上级 23fb0783
展开全部 隐藏空白更改
内联 并排
Showing with 1160 addition and 528 deletion
apm-sniffer/apm-sdk-plugin/grpc-1.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/grpc/v1/BlockingCallClientInterceptor.java 已删除 100644 → 0
浏览文件 @ 23fb0783
/*
* 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.skywalking.apm.plugin.grpc.v1;
import io.grpc.Channel;
import io.grpc.ClientCall;
import io.grpc.ForwardingClientCall;
import io.grpc.Metadata;
import io.grpc.MethodDescriptor;
import org.apache.skywalking.apm.agent.core.context.CarrierItem;
import org.apache.skywalking.apm.agent.core.context.ContextCarrier;
import org.apache.skywalking.apm.agent.core.context.ContextManager;
/**
* @author zhang xin
*/
public class BlockingCallClientInterceptor extends ForwardingClientCall.SimpleForwardingClientCall implements CallClientInterceptor {
private final MethodDescriptor methodDescriptor;
private final Channel channel;
public BlockingCallClientInterceptor(ClientCall delegate, MethodDescriptor method, Channel channel) {
super(delegate);
this.methodDescriptor = method;
this.channel = channel;
}
@Override public void start(Listener responseListener, Metadata headers) {
final ContextCarrier contextCarrier = new ContextCarrier();
ContextManager.inject(contextCarrier);
CarrierItem contextItem = contextCarrier.items();
while (contextItem.hasNext()) {
contextItem = contextItem.next();
Metadata.Key<String> headerKey = Metadata.Key.of(contextItem.getHeadKey(), Metadata.ASCII_STRING_MARSHALLER);
headers.put(headerKey, contextItem.getHeadValue());
}
delegate().start(responseListener, headers);
}
public MethodDescriptor getMethodDescriptor() {
return methodDescriptor;
}
public Channel getChannel() {
return channel;
}
}
apm-sniffer/apm-sdk-plugin/grpc-1.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/grpc/v1/CallServerInterceptor.java 已删除 100644 → 0
浏览文件 @ 23fb0783
/*
* 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.skywalking.apm.plugin.grpc.v1;
import io.grpc.ForwardingServerCall;
import io.grpc.ForwardingServerCallListener;
import io.grpc.Metadata;
import io.grpc.MethodDescriptor;
import io.grpc.ServerCall;
import io.grpc.ServerCallHandler;
import io.grpc.ServerInterceptor;
import java.util.HashMap;
import java.util.Map;
import org.apache.skywalking.apm.agent.core.context.CarrierItem;
import org.apache.skywalking.apm.agent.core.context.ContextCarrier;
import org.apache.skywalking.apm.agent.core.context.ContextManager;
import org.apache.skywalking.apm.agent.core.context.ContextSnapshot;
import org.apache.skywalking.apm.agent.core.context.trace.AbstractSpan;
import org.apache.skywalking.apm.agent.core.context.trace.SpanLayer;
import org.apache.skywalking.apm.network.trace.component.ComponentsDefine;
import org.apache.skywalking.apm.util.StringUtil;
import static org.apache.skywalking.apm.plugin.grpc.v1.Constants.SERVER;
import static org.apache.skywalking.apm.plugin.grpc.v1.Constants.STREAM_REQUEST_OBSERVER_ON_COMPLETE_OPERATION_NAME;
import static org.apache.skywalking.apm.plugin.grpc.v1.Constants.STREAM_REQUEST_OBSERVER_ON_ERROR_OPERATION_NAME;
import static org.apache.skywalking.apm.plugin.grpc.v1.Constants.STREAM_REQUEST_OBSERVER_ON_NEXT_OPERATION_NAME;
/**
* @author zhang xin
*/
public class CallServerInterceptor implements ServerInterceptor {
@Override
public ServerCall.Listener interceptCall(ServerCall call, Metadata headers, ServerCallHandler handler) {
Map<String, String> headerMap = new HashMap<String, String>();
for (String key : headers.keys()) {
if (!key.endsWith(Metadata.BINARY_HEADER_SUFFIX)) {
String value = headers.get(Metadata.Key.of(key, Metadata.ASCII_STRING_MARSHALLER));
headerMap.put(key, value);
}
}
ContextCarrier contextCarrier = new ContextCarrier();
CarrierItem next = contextCarrier.items();
while (next.hasNext()) {
next = next.next();
String contextValue = headerMap.get(next.getHeadKey());
if (!StringUtil.isEmpty(contextValue)) {
next.setHeadValue(contextValue);
}
}
final AbstractSpan span = ContextManager.createEntrySpan(OperationNameFormatUtil.formatOperationName(call.getMethodDescriptor()), contextCarrier);
span.setComponent(ComponentsDefine.GRPC);
SpanLayer.asRPCFramework(span);
try {
return new ServerCallListener(handler.startCall(new ForwardingServerCall.SimpleForwardingServerCall(call) {
@Override
public void sendHeaders(Metadata responseHeaders) {
delegate().sendHeaders(responseHeaders);
}
}, headers), call.getMethodDescriptor(), ContextManager.capture());
} finally {
ContextManager.stopSpan();
}
}
public class ServerCallListener extends ForwardingServerCallListener.SimpleForwardingServerCallListener {
private final ContextSnapshot contextSnapshot;
private final MethodDescriptor.MethodType methodType;
private final String operationPrefix;
protected ServerCallListener(ServerCall.Listener delegate, MethodDescriptor descriptor,
ContextSnapshot contextSnapshot) {
super(delegate);
this.contextSnapshot = contextSnapshot;
this.methodType = descriptor.getType();
this.operationPrefix = OperationNameFormatUtil.formatOperationName(descriptor) + SERVER;
}
@Override public void onReady() {
delegate().onReady();
}
@Override public void onMessage(Object message) {
try {
ContextManager.createLocalSpan(operationPrefix + STREAM_REQUEST_OBSERVER_ON_NEXT_OPERATION_NAME);
ContextManager.continued(contextSnapshot);
delegate().onMessage(message);
} catch (Throwable t) {
ContextManager.activeSpan().errorOccurred().log(t);
} finally {
ContextManager.stopSpan();
}
}
@Override public void onComplete() {
if (methodType != MethodDescriptor.MethodType.UNARY) {
try {
ContextManager.createLocalSpan(operationPrefix + STREAM_REQUEST_OBSERVER_ON_COMPLETE_OPERATION_NAME);
ContextManager.continued(contextSnapshot);
delegate().onComplete();
} catch (Throwable t) {
ContextManager.activeSpan().errorOccurred().log(t);
} finally {
ContextManager.stopSpan();
}
} else {
delegate().onComplete();
}
}
@Override public void onCancel() {
if (methodType != MethodDescriptor.MethodType.UNARY) {
try {
ContextManager.createLocalSpan(operationPrefix + STREAM_REQUEST_OBSERVER_ON_ERROR_OPERATION_NAME);
ContextManager.continued(contextSnapshot);
delegate().onCancel();
} catch (Throwable t) {
ContextManager.activeSpan().errorOccurred().log(t);
} finally {
ContextManager.stopSpan();
}
} else {
delegate().onCancel();
}
}
@Override public void onHalfClose() {
delegate().onHalfClose();
}
}
}
apm-sniffer/apm-sdk-plugin/grpc-1.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/grpc/v1/Constants.java
浏览文件 @ 97950807
......@@ -21,23 +21,52 @@ package org.apache.skywalking.apm.plugin.grpc.v1;
/**
* Constant variables
*
* @author zhang xin
* @author zhang xin, wang zheng, kanro
*/
public class Constants {
/**
* Mark the current application is the gRPC client in current tracing span.
*/
public static final String CLIENT = "/client";
/**
* Mark the current application is the gRPC server in current tracing span.
*/
public static final String SERVER = "/server";
public static final String STREAM_REQUEST_OBSERVER_ON_NEXT_OPERATION_NAME = "/RequestObserver/onNext";
public static final String STREAM_REQUEST_OBSERVER_ON_ERROR_OPERATION_NAME = "/RequestObserver/onError";
/**
* Operation name for request message received on server or sent on client.
*
* Spans of this operations just be create with request stream calls.
*/
public static final String REQUEST_ON_MESSAGE_OPERATION_NAME = "/Request/onMessage";
public static final String STREAM_REQUEST_OBSERVER_ON_COMPLETE_OPERATION_NAME = "/RequestObserver/onComplete";
/**
* Operation name for client has completed request sending, there are no more incoming request.
*
* It should happen with half close state usually.
*/
public static final String REQUEST_ON_COMPLETE_OPERATION_NAME = "/Request/onComplete";
public static final String STREAM_RESPONSE_OBSERVER_ON_NEXT_OPERATION_NAME = "/ResponseObserver/onNext";
/**
* Operation name for client has cancelled the call.
*/
public static final String REQUEST_ON_CANCEL_OPERATION_NAME = "/Request/onCancel";
public static final String STREAM_RESPONSE_OBSERVER_ON_ERROR_OPERATION_NAME = "/ResponseObserver/onError";
/**
* Operation name for response message received on client or sent on server.
*
* Spans of this operations just be create with response stream calls.
*/
public static final String RESPONSE_ON_MESSAGE_OPERATION_NAME = "/Response/onMessage";
public static final String STREAM_RESPONSE_OBSERVER_ON_COMPLETE_OPERATION_NAME = "/ResponseObserver/onComplete";
/**
* Operation name for call closed with status and trailers.
*
* Exceptions will be logs here.
*/
public static final String RESPONSE_ON_CLOSE_OPERATION_NAME = "/Response/onClose";
}
public static final String BLOCKING_CALL_EXIT_SPAN = "SW_GRPC_BLOCKING_CALL_EXIT_SPAN";
}
\ No newline at end of file
apm-sniffer/apm-sdk-plugin/grpc-1.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/grpc/v1/OperationNameFormatUtil.java
浏览文件 @ 97950807
......@@ -27,7 +27,7 @@ import io.grpc.MethodDescriptor;
*/
public class OperationNameFormatUtil {
public static String formatOperationName(MethodDescriptor methodDescriptor) {
public static String formatOperationName(MethodDescriptor<?, ?> methodDescriptor) {
String fullMethodName = methodDescriptor.getFullMethodName();
return formatServiceName(fullMethodName) + "." + formatMethodName(fullMethodName);
}
......
apm-sniffer/apm-sdk-plugin/grpc-1.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/grpc/v1/AbstractStubInterceptor.java → apm-sniffer/apm-sdk-plugin/grpc-1.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/grpc/v1/client/AbstractStubInterceptor.java
浏览文件 @ 97950807
......@@ -16,40 +16,42 @@
*
*/
package org.apache.skywalking.apm.plugin.grpc.v1;
package org.apache.skywalking.apm.plugin.grpc.v1.client;
import io.grpc.Channel;
import io.grpc.ClientInterceptors;
import java.lang.reflect.Method;
import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.EnhancedInstance;
import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.InstanceConstructorInterceptor;
import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.InstanceMethodsAroundInterceptor;
import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.MethodInterceptResult;
import java.lang.reflect.Method;
/**
* {@link AbstractStubInterceptor} add the interceptor for every ClientCall.
*
* @author zhang xin
* @author zhang xin, kanro
*/
public class AbstractStubInterceptor implements InstanceMethodsAroundInterceptor, InstanceConstructorInterceptor {
@Override
public void onConstruct(EnhancedInstance objInst, Object[] allArguments) {
Channel channel = (Channel)allArguments[0];
objInst.setSkyWalkingDynamicField(ClientInterceptors.intercept(channel, new GRPCClientInterceptor()));
Channel channel = (Channel) allArguments[0];
objInst.setSkyWalkingDynamicField(ClientInterceptors.intercept(channel, new ClientInterceptor()));
}
@Override
public void beforeMethod(EnhancedInstance objInst, Method method, Object[] allArguments, Class<?>[] argumentsTypes,
MethodInterceptResult result) throws Throwable {
MethodInterceptResult result) throws Throwable {
}
@Override
public Object afterMethod(EnhancedInstance objInst, Method method, Object[] allArguments, Class<?>[] argumentsTypes,
Object ret) throws Throwable {
Object ret) throws Throwable {
return objInst.getSkyWalkingDynamicField();
}
@Override public void handleMethodException(EnhancedInstance objInst, Method method, Object[] allArguments,
Class<?>[] argumentsTypes, Throwable t) {
@Override
public void handleMethodException(EnhancedInstance objInst, Method method, Object[] allArguments,
Class<?>[] argumentsTypes, Throwable t) {
}
}
apm-sniffer/apm-sdk-plugin/grpc-1.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/grpc/v1/BlockingCallInterceptor.java → apm-sniffer/apm-sdk-plugin/grpc-1.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/grpc/v1/client/BlockingCallInterceptor.java
浏览文件 @ 97950807
......@@ -16,11 +16,10 @@
*
*/
package org.apache.skywalking.apm.plugin.grpc.v1;
package org.apache.skywalking.apm.plugin.grpc.v1.client;
import io.grpc.Channel;
import io.grpc.MethodDescriptor;
import java.lang.reflect.Method;
import org.apache.skywalking.apm.agent.core.context.ContextManager;
import org.apache.skywalking.apm.agent.core.context.trace.AbstractSpan;
import org.apache.skywalking.apm.agent.core.context.trace.SpanLayer;
......@@ -28,31 +27,37 @@ import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.MethodInt
import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.StaticMethodsAroundInterceptor;
import org.apache.skywalking.apm.network.trace.component.ComponentsDefine;
import java.lang.reflect.Method;
import static org.apache.skywalking.apm.plugin.grpc.v1.Constants.BLOCKING_CALL_EXIT_SPAN;
import static org.apache.skywalking.apm.plugin.grpc.v1.OperationNameFormatUtil.formatOperationName;
/**
* @author zhang xin
* @author zhang xin, kanro
*/
public class BlockingCallInterceptor implements StaticMethodsAroundInterceptor {
@Override public void beforeMethod(Class clazz, Method method, Object[] allArguments, Class<?>[] parameterTypes,
MethodInterceptResult result) {
Channel channel = (Channel)allArguments[0];
MethodDescriptor methodDescriptor = (MethodDescriptor)allArguments[1];
@Override
public void beforeMethod(Class clazz, Method method, Object[] allArguments, Class<?>[] parameterTypes,
MethodInterceptResult result) {
Channel channel = (Channel) allArguments[0];
MethodDescriptor<?, ?> methodDescriptor = (MethodDescriptor<?, ?>) allArguments[1];
final AbstractSpan span = ContextManager.createExitSpan(formatOperationName(methodDescriptor), channel.authority());
span.setComponent(ComponentsDefine.GRPC);
SpanLayer.asRPCFramework(span);
span.setLayer(SpanLayer.RPC_FRAMEWORK);
ContextManager.getRuntimeContext().put(BLOCKING_CALL_EXIT_SPAN, span);
}
@Override public Object afterMethod(Class clazz, Method method, Object[] allArguments, Class<?>[] parameterTypes,
Object ret) {
@Override
public Object afterMethod(Class clazz, Method method, Object[] allArguments, Class<?>[] parameterTypes,
Object ret) {
ContextManager.stopSpan();
return ret;
}
@Override
public void handleMethodException(Class clazz, Method method, Object[] allArguments, Class<?>[] parameterTypes,
Throwable t) {
Throwable t) {
ContextManager.activeSpan().errorOccurred().log(t);
}
}
apm-sniffer/apm-sdk-plugin/grpc-1.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/grpc/v1/CallClientInterceptor.java → apm-sniffer/apm-sdk-plugin/grpc-1.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/grpc/v1/client/ClientInterceptor.java
浏览文件 @ 97950807
......@@ -16,19 +16,21 @@
*
*/
package org.apache.skywalking.apm.plugin.grpc.v1;
package org.apache.skywalking.apm.plugin.grpc.v1.client;
import io.grpc.CallOptions;
import io.grpc.Channel;
import io.grpc.ClientCall;
import io.grpc.MethodDescriptor;
/**
* @author AI
* 2019-07-22
* @author zhang xin, kanro
*/
public interface CallClientInterceptor {
public Channel getChannel();
public MethodDescriptor getMethodDescriptor();
public class ClientInterceptor implements io.grpc.ClientInterceptor {
@Override
public <REQUEST, RESPONSE> ClientCall<REQUEST, RESPONSE> interceptCall(MethodDescriptor<REQUEST, RESPONSE> method,
CallOptions callOptions, Channel channel) {
return new TracingClientCall<>(channel.newCall(method, callOptions), method, channel);
}
}
apm-sniffer/apm-sdk-plugin/grpc-1.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/grpc/v1/StreamCallClientInterceptor.java → apm-sniffer/apm-sdk-plugin/grpc-1.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/grpc/v1/client/TracingClientCall.java
浏览文件 @ 97950807
......@@ -16,15 +16,9 @@
*
*/
package org.apache.skywalking.apm.plugin.grpc.v1;
import io.grpc.Channel;
import io.grpc.ClientCall;
import io.grpc.ForwardingClientCall;
import io.grpc.ForwardingClientCallListener;
import io.grpc.Metadata;
import io.grpc.MethodDescriptor;
import io.grpc.Status;
package org.apache.skywalking.apm.plugin.grpc.v1.client;
import io.grpc.*;
import org.apache.skywalking.apm.agent.core.context.CarrierItem;
import org.apache.skywalking.apm.agent.core.context.ContextCarrier;
import org.apache.skywalking.apm.agent.core.context.ContextManager;
......@@ -33,27 +27,29 @@ import org.apache.skywalking.apm.agent.core.context.tag.Tags;
import org.apache.skywalking.apm.agent.core.context.trace.AbstractSpan;
import org.apache.skywalking.apm.agent.core.context.trace.SpanLayer;
import org.apache.skywalking.apm.network.trace.component.ComponentsDefine;
import org.apache.skywalking.apm.plugin.grpc.v1.OperationNameFormatUtil;
import javax.annotation.Nullable;
import static org.apache.skywalking.apm.plugin.grpc.v1.Constants.CLIENT;
import static org.apache.skywalking.apm.plugin.grpc.v1.Constants.STREAM_RESPONSE_OBSERVER_ON_COMPLETE_OPERATION_NAME;
import static org.apache.skywalking.apm.plugin.grpc.v1.Constants.STREAM_RESPONSE_OBSERVER_ON_ERROR_OPERATION_NAME;
import static org.apache.skywalking.apm.plugin.grpc.v1.Constants.STREAM_RESPONSE_OBSERVER_ON_NEXT_OPERATION_NAME;
import static org.apache.skywalking.apm.plugin.grpc.v1.Constants.*;
import static org.apache.skywalking.apm.plugin.grpc.v1.OperationNameFormatUtil.formatOperationName;
/**
* @author zhangxin
* Fully client tracing for gRPC servers.
*
* @author zhang xin, kanro
*/
public class StreamCallClientInterceptor extends ForwardingClientCall.SimpleForwardingClientCall implements CallClientInterceptor {
class TracingClientCall<REQUEST, RESPONSE> extends ForwardingClientCall.SimpleForwardingClientCall<REQUEST, RESPONSE> {
private final String serviceName;
private final String remotePeer;
private final String operationPrefix;
private final Channel channel;
private final MethodDescriptor methodDescriptor;
private final MethodDescriptor<REQUEST, RESPONSE> methodDescriptor;
private ContextSnapshot snapshot;
protected StreamCallClientInterceptor(ClientCall delegate, MethodDescriptor method, Channel channel) {
TracingClientCall(ClientCall<REQUEST, RESPONSE> delegate, MethodDescriptor<REQUEST, RESPONSE> method, Channel channel) {
super(delegate);
this.channel = channel;
this.methodDescriptor = method;
this.serviceName = formatOperationName(method);
this.remotePeer = channel.authority();
......@@ -61,55 +57,121 @@ public class StreamCallClientInterceptor extends ForwardingClientCall.SimpleForw
}
@Override
public void start(Listener responseListener, Metadata headers) {
public void start(Listener<RESPONSE> responseListener, Metadata headers) {
final AbstractSpan blockingSpan = (AbstractSpan) ContextManager.getRuntimeContext().get(BLOCKING_CALL_EXIT_SPAN);
final ContextCarrier contextCarrier = new ContextCarrier();
final AbstractSpan span = ContextManager.createExitSpan(serviceName, contextCarrier, remotePeer);
span.setComponent(ComponentsDefine.GRPC);
SpanLayer.asRPCFramework(span);
// Avoid create ExitSpan repeatedly, ExitSpan of blocking calls will create by BlockingCallInterceptor.
if (blockingSpan == null) {
final AbstractSpan span = ContextManager.createExitSpan(serviceName, remotePeer);
span.setComponent(ComponentsDefine.GRPC);
span.setLayer(SpanLayer.RPC_FRAMEWORK);
} else {
ContextManager.getRuntimeContext().remove(BLOCKING_CALL_EXIT_SPAN);
}
ContextManager.inject(contextCarrier);
CarrierItem contextItem = contextCarrier.items();
while (contextItem.hasNext()) {
contextItem = contextItem.next();
Metadata.Key<String> headerKey = Metadata.Key.of(contextItem.getHeadKey(), Metadata.ASCII_STRING_MARSHALLER);
headers.put(headerKey, contextItem.getHeadValue());
}
delegate().start(new CallListener(responseListener, ContextManager.capture()), headers);
ContextManager.stopSpan();
snapshot = ContextManager.capture();
try {
delegate().start(new TracingClientCallListener(responseListener, snapshot), headers);
} catch (Throwable t) {
ContextManager.activeSpan().errorOccurred().log(t);
throw t;
} finally {
if (blockingSpan == null) {
ContextManager.stopSpan();
}
}
}
@Override
public Channel getChannel() {
return channel;
public void sendMessage(REQUEST message) {
if (methodDescriptor.getType().clientSendsOneMessage()) {
super.sendMessage(message);
return;
}
final AbstractSpan span = ContextManager.createLocalSpan(operationPrefix + REQUEST_ON_MESSAGE_OPERATION_NAME);
span.setComponent(ComponentsDefine.GRPC);
span.setLayer(SpanLayer.RPC_FRAMEWORK);
ContextManager.continued(snapshot);
try {
super.sendMessage(message);
} catch (Throwable t) {
ContextManager.activeSpan().errorOccurred().log(t);
throw t;
} finally {
ContextManager.stopSpan();
}
}
@Override
public MethodDescriptor getMethodDescriptor() {
return methodDescriptor;
public void halfClose() {
final AbstractSpan span = ContextManager.createLocalSpan(operationPrefix + REQUEST_ON_COMPLETE_OPERATION_NAME);
span.setComponent(ComponentsDefine.GRPC);
span.setLayer(SpanLayer.RPC_FRAMEWORK);
ContextManager.continued(snapshot);
try {
super.halfClose();
} catch (Throwable t) {
ContextManager.activeSpan().errorOccurred().log(t);
throw t;
} finally {
ContextManager.stopSpan();
}
}
private class CallListener extends ForwardingClientCallListener.SimpleForwardingClientCallListener {
@Override
public void cancel(@Nullable String message, @Nullable Throwable cause) {
final AbstractSpan span = ContextManager.createLocalSpan(operationPrefix + REQUEST_ON_CANCEL_OPERATION_NAME);
span.setComponent(ComponentsDefine.GRPC);
span.setLayer(SpanLayer.RPC_FRAMEWORK);
ContextManager.continued(snapshot);
if (cause != null) {
span.log(cause);
}
try {
super.cancel(message, cause);
} catch (Throwable t) {
ContextManager.activeSpan().errorOccurred().log(t);
throw t;
} finally {
ContextManager.stopSpan();
}
}
class TracingClientCallListener extends ForwardingClientCallListener.SimpleForwardingClientCallListener<RESPONSE> {
private final ContextSnapshot contextSnapshot;
protected CallListener(Listener delegate, ContextSnapshot contextSnapshot) {
TracingClientCallListener(Listener<RESPONSE> delegate, ContextSnapshot contextSnapshot) {
super(delegate);
this.contextSnapshot = contextSnapshot;
}
@Override
public void onReady() {
delegate().onReady();
}
public void onMessage(RESPONSE message) {
if (methodDescriptor.getType().serverSendsOneMessage()) {
super.onMessage(message);
return;
}
@Override
public void onHeaders(Metadata headers) {
delegate().onHeaders(headers);
}
final AbstractSpan span = ContextManager.createLocalSpan(operationPrefix + RESPONSE_ON_MESSAGE_OPERATION_NAME);
span.setComponent(ComponentsDefine.GRPC);
span.setLayer(SpanLayer.RPC_FRAMEWORK);
ContextManager.continued(contextSnapshot);
@Override
public void onMessage(Object message) {
try {
ContextManager.createLocalSpan(operationPrefix + STREAM_RESPONSE_OBSERVER_ON_NEXT_OPERATION_NAME);
ContextManager.continued(contextSnapshot);
delegate().onMessage(message);
} catch (Throwable t) {
ContextManager.activeSpan().errorOccurred().log(t);
......@@ -120,16 +182,17 @@ public class StreamCallClientInterceptor extends ForwardingClientCall.SimpleForw
@Override
public void onClose(Status status, Metadata trailers) {
final AbstractSpan span = ContextManager.createLocalSpan(operationPrefix + RESPONSE_ON_CLOSE_OPERATION_NAME);
span.setComponent(ComponentsDefine.GRPC);
span.setLayer(SpanLayer.RPC_FRAMEWORK);
ContextManager.continued(contextSnapshot);
if (!status.isOk()) {
span.errorOccurred().log(status.asRuntimeException());
Tags.STATUS_CODE.set(span, status.getCode().name());
}
try {
if (!status.isOk()) {
AbstractSpan abstractSpan = ContextManager.createLocalSpan(operationPrefix + STREAM_RESPONSE_OBSERVER_ON_ERROR_OPERATION_NAME);
abstractSpan.errorOccurred().log(status.asRuntimeException());
Tags.STATUS_CODE.set(abstractSpan, status.getCode().name());
} else {
AbstractSpan abstractSpan = ContextManager.createLocalSpan(operationPrefix + STREAM_RESPONSE_OBSERVER_ON_COMPLETE_OPERATION_NAME);
}
delegate().onClose(status, trailers);
ContextManager.continued(contextSnapshot);
} catch (Throwable t) {
ContextManager.activeSpan().errorOccurred().log(t);
} finally {
......@@ -137,5 +200,4 @@ public class StreamCallClientInterceptor extends ForwardingClientCall.SimpleForw
}
}
}
}
apm-sniffer/apm-sdk-plugin/grpc-1.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/grpc/v1/define/AbstractServerImplBuilderInstrumentation.java
浏览文件 @ 97950807
......@@ -33,7 +33,7 @@ public class AbstractServerImplBuilderInstrumentation extends ClassInstanceMetho
public static final String ENHANCE_CLASS = "io.grpc.internal.AbstractServerImplBuilder";
public static final String ENHANCE_METHOD = "addService";
public static final String INTERCEPT_CLASS = "org.apache.skywalking.apm.plugin.grpc.v1.AbstractServerImplBuilderInterceptor";
public static final String INTERCEPT_CLASS = "org.apache.skywalking.apm.plugin.grpc.v1.server.AbstractServerImplBuilderInterceptor";
public static final String ARGUMENT_TYPE = "io.grpc.ServerServiceDefinition";
@Override public ConstructorInterceptPoint[] getConstructorsInterceptPoints() {
......
apm-sniffer/apm-sdk-plugin/grpc-1.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/grpc/v1/define/AbstractStubInstrumentation.java
浏览文件 @ 97950807
......@@ -31,7 +31,7 @@ import static org.apache.skywalking.apm.agent.core.plugin.match.NameMatch.byName
public class AbstractStubInstrumentation extends ClassInstanceMethodsEnhancePluginDefine {
public static final String INTERCEPT_CLASS = "org.apache.skywalking.apm.plugin.grpc.v1.AbstractStubInterceptor";
public static final String INTERCEPT_CLASS = "org.apache.skywalking.apm.plugin.grpc.v1.client.AbstractStubInterceptor";
public static final String ENHANCE_METHOD = "getChannel";
public static final String ENHANCE_CLASS = "io.grpc.stub.AbstractStub";
......
apm-sniffer/apm-sdk-plugin/grpc-1.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/grpc/v1/define/ClientCallsInstrumentation.java
浏览文件 @ 97950807
......@@ -32,10 +32,8 @@ import static org.apache.skywalking.apm.agent.core.plugin.match.NameMatch.byName
* @author zhang xin
*/
public class ClientCallsInstrumentation extends ClassStaticMethodsEnhancePluginDefine {
private static final String ENHANCE_CLASS = "io.grpc.stub.ClientCalls";
private static final String INTERCEPTOR_CLASS = "org.apache.skywalking.apm.plugin.grpc.v1.BlockingCallInterceptor";
private static final String FUTURE_INTERCEPTOR_CLASS = "org.apache.skywalking.apm.plugin.grpc.v1.AsyncUnaryRequestCallCallInterceptor";
private static final String INTERCEPTOR_CLASS = "org.apache.skywalking.apm.plugin.grpc.v1.client.BlockingCallInterceptor";
@Override public StaticMethodsInterceptPoint[] getStaticMethodsInterceptPoints() {
return new StaticMethodsInterceptPoint[] {
......@@ -51,32 +49,6 @@ public class ClientCallsInstrumentation extends ClassStaticMethodsEnhancePluginD
@Override public boolean isOverrideArgs() {
return false;
}
},
// new StaticMethodsInterceptPoint() {
// @Override public ElementMatcher<MethodDescription> getMethodsMatcher() {
// return named("blockingServerStreamingCall").and(takesArgumentWithType(1, "io.grpc.MethodDescriptor"));
// }
//
// @Override public String getMethodsInterceptor() {
// return INTERCEPTOR_CLASS;
// }
//
// @Override public boolean isOverrideArgs() {
// return false;
// }
// },
new StaticMethodsInterceptPoint() {
@Override public ElementMatcher<MethodDescription> getMethodsMatcher() {
return named("asyncUnaryRequestCall").and(takesArgumentWithType(2, "io.grpc.ClientCall$Listener"));
}
@Override public String getMethodsInterceptor() {
return FUTURE_INTERCEPTOR_CLASS;
}
@Override public boolean isOverrideArgs() {
return true;
}
}
};
}
......
apm-sniffer/apm-sdk-plugin/grpc-1.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/grpc/v1/AbstractServerImplBuilderInterceptor.java → apm-sniffer/apm-sdk-plugin/grpc-1.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/grpc/v1/server/AbstractServerImplBuilderInterceptor.java
浏览文件 @ 97950807
......@@ -16,17 +16,19 @@
*
*/
package org.apache.skywalking.apm.plugin.grpc.v1;
package org.apache.skywalking.apm.plugin.grpc.v1.server;
import io.grpc.ServerInterceptors;
import io.grpc.ServerServiceDefinition;
import java.lang.reflect.Method;
import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.EnhancedInstance;
import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.InstanceMethodsAroundInterceptor;
import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.MethodInterceptResult;
/**
* {@link AbstractServerImplBuilderInterceptor} add the {@link CallServerInterceptor} interceptor for every
* {@link AbstractServerImplBuilderInterceptor} add the {@link ServerInterceptor} interceptor for every
* ServerService.
*
* @author zhang xin
......@@ -34,18 +36,19 @@ import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.MethodInt
public class AbstractServerImplBuilderInterceptor implements InstanceMethodsAroundInterceptor {
@Override
public void beforeMethod(EnhancedInstance objInst, Method method, Object[] allArguments, Class<?>[] argumentsTypes,
MethodInterceptResult result) throws Throwable {
allArguments[0] = ServerInterceptors.intercept((ServerServiceDefinition)allArguments[0], new CallServerInterceptor());
MethodInterceptResult result) {
allArguments[0] = ServerInterceptors.intercept((ServerServiceDefinition) allArguments[0], new ServerInterceptor());
}
@Override
public Object afterMethod(EnhancedInstance objInst, Method method, Object[] allArguments, Class<?>[] argumentsTypes,
Object ret) throws Throwable {
Object ret) {
return ret;
}
@Override public void handleMethodException(EnhancedInstance objInst, Method method, Object[] allArguments,
Class<?>[] argumentsTypes, Throwable t) {
@Override
public void handleMethodException(EnhancedInstance objInst, Method method, Object[] allArguments,
Class<?>[] argumentsTypes, Throwable t) {
}
}
apm-sniffer/apm-sdk-plugin/grpc-1.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/grpc/v1/AsyncUnaryRequestCallCallInterceptor.java → apm-sniffer/apm-sdk-plugin/grpc-1.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/grpc/v1/server/ServerInterceptor.java
浏览文件 @ 97950807
......@@ -16,44 +16,43 @@
*
*/
package org.apache.skywalking.apm.plugin.grpc.v1;
package org.apache.skywalking.apm.plugin.grpc.v1.server;
import io.grpc.Channel;
import io.grpc.MethodDescriptor;
import java.lang.reflect.Method;
import io.grpc.Metadata;
import io.grpc.ServerCall;
import io.grpc.ServerCallHandler;
import org.apache.skywalking.apm.agent.core.context.CarrierItem;
import org.apache.skywalking.apm.agent.core.context.ContextCarrier;
import org.apache.skywalking.apm.agent.core.context.ContextManager;
import org.apache.skywalking.apm.agent.core.context.trace.AbstractSpan;
import org.apache.skywalking.apm.agent.core.context.trace.SpanLayer;
import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.MethodInterceptResult;
import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.StaticMethodsAroundInterceptor;
import org.apache.skywalking.apm.network.trace.component.ComponentsDefine;
import static org.apache.skywalking.apm.plugin.grpc.v1.OperationNameFormatUtil.formatOperationName;
import org.apache.skywalking.apm.plugin.grpc.v1.OperationNameFormatUtil;
import org.apache.skywalking.apm.util.StringUtil;
/**
* @author zhang xin
* @author zhang xin, wang zheng, kanro
*/
public class AsyncUnaryRequestCallCallInterceptor implements StaticMethodsAroundInterceptor {
@Override public void beforeMethod(Class clazz, Method method, Object[] allArguments, Class<?>[] parameterTypes,
MethodInterceptResult result) {
CallClientInterceptor originClientCall = (CallClientInterceptor)allArguments[0];
Channel channel = originClientCall.getChannel();
MethodDescriptor methodDescriptor = originClientCall.getMethodDescriptor();
public class ServerInterceptor implements io.grpc.ServerInterceptor {
@Override
public <REQUEST, RESPONSE> ServerCall.Listener<REQUEST> interceptCall(ServerCall<REQUEST, RESPONSE> call, Metadata headers, ServerCallHandler<REQUEST, RESPONSE> handler) {
final ContextCarrier contextCarrier = new ContextCarrier();
CarrierItem next = contextCarrier.items();
while (next.hasNext()) {
next = next.next();
String contextValue = headers.get(Metadata.Key.of(next.getHeadKey(), Metadata.ASCII_STRING_MARSHALLER));
if (!StringUtil.isEmpty(contextValue)) {
next.setHeadValue(contextValue);
}
}
final AbstractSpan span = ContextManager.createExitSpan(formatOperationName(methodDescriptor), channel.authority());
final AbstractSpan span = ContextManager.createEntrySpan(OperationNameFormatUtil.formatOperationName(call.getMethodDescriptor()), contextCarrier);
span.setComponent(ComponentsDefine.GRPC);
SpanLayer.asRPCFramework(span);
}
@Override public Object afterMethod(Class clazz, Method method, Object[] allArguments, Class<?>[] parameterTypes,
Object ret) {
ContextManager.stopSpan();
return ret;
}
@Override
public void handleMethodException(Class clazz, Method method, Object[] allArguments, Class<?>[] parameterTypes,
Throwable t) {
ContextManager.activeSpan().errorOccurred().log(t);
span.setLayer(SpanLayer.RPC_FRAMEWORK);
try {
return new TracingServerCallListener<>(handler.startCall(new TracingServerCall<>(call, ContextManager.capture()), headers), call.getMethodDescriptor(), ContextManager.capture());
} finally {
ContextManager.stopSpan();
}
}
}
apm-sniffer/apm-sdk-plugin/grpc-1.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/grpc/v1/server/TracingServerCall.java 0 → 100644
浏览文件 @ 97950807
/*
* 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.skywalking.apm.plugin.grpc.v1.server;
import io.grpc.ForwardingServerCall;
import io.grpc.Metadata;
import io.grpc.ServerCall;
import io.grpc.Status;
import org.apache.skywalking.apm.agent.core.context.ContextManager;
import org.apache.skywalking.apm.agent.core.context.ContextSnapshot;
import org.apache.skywalking.apm.agent.core.context.tag.Tags;
import org.apache.skywalking.apm.agent.core.context.trace.AbstractSpan;
import org.apache.skywalking.apm.agent.core.context.trace.SpanLayer;
import org.apache.skywalking.apm.network.trace.component.ComponentsDefine;
import org.apache.skywalking.apm.plugin.grpc.v1.OperationNameFormatUtil;
import static org.apache.skywalking.apm.plugin.grpc.v1.Constants.*;
/**
* @author wang zheng, kanro
*/
public class TracingServerCall<REQUEST, RESPONSE> extends ForwardingServerCall.SimpleForwardingServerCall<REQUEST, RESPONSE> {
private final String operationPrefix;
private final ContextSnapshot contextSnapshot;
protected TracingServerCall(ServerCall<REQUEST, RESPONSE> delegate, ContextSnapshot contextSnapshot) {
super(delegate);
this.operationPrefix = OperationNameFormatUtil.formatOperationName(delegate.getMethodDescriptor()) + SERVER;
this.contextSnapshot = contextSnapshot;
}
@Override
public void sendMessage(RESPONSE message) {
// We just create the request on message span for server stream calls.
if (!getMethodDescriptor().getType().serverSendsOneMessage()) {
final AbstractSpan span = ContextManager.createLocalSpan(operationPrefix + RESPONSE_ON_MESSAGE_OPERATION_NAME);
span.setComponent(ComponentsDefine.GRPC);
span.setLayer(SpanLayer.RPC_FRAMEWORK);
ContextManager.continued(contextSnapshot);
try {
super.sendMessage(message);
} catch (Throwable t) {
ContextManager.activeSpan().errorOccurred().log(t);
throw t;
} finally {
ContextManager.stopSpan();
}
} else {
super.sendMessage(message);
}
}
@Override
public void close(Status status, Metadata trailers) {
final AbstractSpan span = ContextManager.createLocalSpan(operationPrefix + RESPONSE_ON_CLOSE_OPERATION_NAME);
span.setComponent(ComponentsDefine.GRPC);
span.setLayer(SpanLayer.RPC_FRAMEWORK);
ContextManager.continued(contextSnapshot);
switch (status.getCode()) {
case OK:
break;
// UNKNOWN/INTERNAL status code will case error in this span.
// Those status code means some unexpected error occurred in server.
// Similar to 5XX in HTTP status.
case UNKNOWN:
case INTERNAL:
if (status.getCause() == null) {
span.errorOccurred().log(status.asRuntimeException());
} else {
span.errorOccurred().log(status.getCause());
}
break;
// Other status code means some predictable error occurred in server.
// Like PERMISSION_DENIED or UNAUTHENTICATED somethings.
// Similar to 4XX in HTTP status.
default:
// But if the status still has cause exception, we will log it too.
if (status.getCause() != null) {
span.errorOccurred().log(status.getCause());
}
break;
}
Tags.STATUS_CODE.set(span, status.getCode().name());
try {
super.close(status, trailers);
} catch (Throwable t) {
ContextManager.activeSpan().errorOccurred().log(t);
throw t;
} finally {
ContextManager.stopSpan();
}
}
}
apm-sniffer/apm-sdk-plugin/grpc-1.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/grpc/v1/server/TracingServerCallListener.java 0 → 100644
浏览文件 @ 97950807
/*
* 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.skywalking.apm.plugin.grpc.v1.server;
import io.grpc.ForwardingServerCallListener;
import io.grpc.MethodDescriptor;
import io.grpc.ServerCall;
import org.apache.skywalking.apm.agent.core.context.ContextManager;
import org.apache.skywalking.apm.agent.core.context.ContextSnapshot;
import org.apache.skywalking.apm.agent.core.context.trace.AbstractSpan;
import org.apache.skywalking.apm.agent.core.context.trace.SpanLayer;
import org.apache.skywalking.apm.network.trace.component.ComponentsDefine;
import org.apache.skywalking.apm.plugin.grpc.v1.OperationNameFormatUtil;
import static org.apache.skywalking.apm.plugin.grpc.v1.Constants.*;
/**
* @author wang zheng, kanro
*/
public class TracingServerCallListener<REQUEST> extends ForwardingServerCallListener.SimpleForwardingServerCallListener<REQUEST> {
private final ContextSnapshot contextSnapshot;
private final MethodDescriptor.MethodType methodType;
private final String operationPrefix;
protected TracingServerCallListener(ServerCall.Listener<REQUEST> delegate, MethodDescriptor<REQUEST, ?> descriptor,
ContextSnapshot contextSnapshot) {
super(delegate);
this.contextSnapshot = contextSnapshot;
this.methodType = descriptor.getType();
this.operationPrefix = OperationNameFormatUtil.formatOperationName(descriptor) + SERVER;
}
@Override
public void onMessage(REQUEST message) {
// We just create the request on message span for client stream calls.
if (!methodType.clientSendsOneMessage()) {
final AbstractSpan span = ContextManager.createLocalSpan(operationPrefix + REQUEST_ON_MESSAGE_OPERATION_NAME);
span.setComponent(ComponentsDefine.GRPC);
span.setLayer(SpanLayer.RPC_FRAMEWORK);
ContextManager.continued(contextSnapshot);
try {
super.onMessage(message);
} catch (Throwable t) {
ContextManager.activeSpan().errorOccurred().log(t);
throw t;
} finally {
ContextManager.stopSpan();
}
} else {
super.onMessage(message);
}
}
@Override
public void onCancel() {
final AbstractSpan span = ContextManager.createLocalSpan(operationPrefix + REQUEST_ON_CANCEL_OPERATION_NAME);
span.setComponent(ComponentsDefine.GRPC);
span.setLayer(SpanLayer.RPC_FRAMEWORK);
ContextManager.continued(contextSnapshot);
try {
super.onCancel();
} catch (Throwable t) {
ContextManager.activeSpan().errorOccurred().log(t);
throw t;
} finally {
ContextManager.stopSpan();
}
}
@Override
public void onHalfClose() {
final AbstractSpan span = ContextManager.createLocalSpan(operationPrefix + REQUEST_ON_COMPLETE_OPERATION_NAME);
span.setComponent(ComponentsDefine.GRPC);
span.setLayer(SpanLayer.RPC_FRAMEWORK);
ContextManager.continued(contextSnapshot);
try {
super.onHalfClose();
} catch (Throwable t) {
ContextManager.activeSpan().errorOccurred().log(t);
throw t;
} finally {
ContextManager.stopSpan();
}
}
}
test/plugin/scenarios/grpc-scenario/src/main/java/org/apache/skywalking/apm/testcase/grpc/controller/CaseController.java
浏览文件 @ 97950807
......@@ -26,10 +26,7 @@ import io.grpc.stub.ClientResponseObserver;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.apache.skywalking.apm.testcase.grpc.consumr.ConsumerInterceptor;
import org.apache.skywalking.apm.testcase.grpc.proto.GreeterBlockingGrpc;
import org.apache.skywalking.apm.testcase.grpc.proto.GreeterGrpc;
import org.apache.skywalking.apm.testcase.grpc.proto.HelloReply;
import org.apache.skywalking.apm.testcase.grpc.proto.HelloRequest;
import org.apache.skywalking.apm.testcase.grpc.proto.*;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.ResponseBody;
import org.springframework.web.bind.annotation.RestController;
......@@ -52,12 +49,14 @@ public class CaseController {
private ManagedChannel channel;
private GreeterGrpc.GreeterStub greeterStub;
private GreeterBlockingGrpc.GreeterBlockingBlockingStub greeterBlockingStub;
private GreeterBlockingErrorGrpc.GreeterBlockingErrorBlockingStub greeterBlockingErrorStub;
@PostConstruct
public void up() {
channel = ManagedChannelBuilder.forAddress(gprcProviderHost, grpcProviderPort).usePlaintext(true).build();
greeterStub = GreeterGrpc.newStub(ClientInterceptors.intercept(channel, new ConsumerInterceptor()));
greeterBlockingStub = GreeterBlockingGrpc.newBlockingStub(ClientInterceptors.intercept(channel, new ConsumerInterceptor()));
greeterBlockingErrorStub = GreeterBlockingErrorGrpc.newBlockingStub(ClientInterceptors.intercept(channel, new ConsumerInterceptor()));
}
@RequestMapping("/grpc-scenario")
......@@ -65,6 +64,7 @@ public class CaseController {
public String testcase() {
greetService();
greetBlockingService();
greetBlockingErrorService();
return SUCCESS;
}
......@@ -129,4 +129,9 @@ public class CaseController {
HelloRequest request = HelloRequest.newBuilder().setName("Sophia").build();
greeterBlockingStub.sayHello(request);
}
private void greetBlockingErrorService() {
HelloRequest request = HelloRequest.newBuilder().setName("Tony").build();
greeterBlockingErrorStub.sayHello(request);
}
}
test/plugin/scenarios/grpc-scenario/src/main/java/org/apache/skywalking/apm/testcase/grpc/provider/ProviderConfiguration.java
浏览文件 @ 97950807
......@@ -22,6 +22,7 @@ import io.grpc.Server;
import io.grpc.ServerBuilder;
import io.grpc.ServerInterceptors;
import org.apache.skywalking.apm.testcase.grpc.provider.interceptor.ProviderInterceptor;
import org.apache.skywalking.apm.testcase.grpc.provider.service.GreeterBlockingErrorServiceImpl;
import org.apache.skywalking.apm.testcase.grpc.provider.service.GreeterBlockingServiceImpl;
import org.apache.skywalking.apm.testcase.grpc.provider.service.GreeterServiceImpl;
import org.springframework.beans.factory.annotation.Configurable;
......@@ -38,8 +39,9 @@ public class ProviderConfiguration {
@Bean(initMethod = "start", destroyMethod = "shutdown")
public Server server() {
return ServerBuilder.forPort(18080)
.addService(ServerInterceptors.intercept(new GreeterServiceImpl(), new ProviderInterceptor()))
.addService(ServerInterceptors.intercept(new GreeterBlockingServiceImpl(), new ProviderInterceptor()))
.build();
.addService(ServerInterceptors.intercept(new GreeterServiceImpl(), new ProviderInterceptor()))
.addService(ServerInterceptors.intercept(new GreeterBlockingServiceImpl(), new ProviderInterceptor()))
.addService(ServerInterceptors.intercept(new GreeterBlockingErrorServiceImpl(), new ProviderInterceptor()))
.build();
}
}
apm-sniffer/apm-sdk-plugin/grpc-1.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/grpc/v1/GRPCClientInterceptor.java → test/plugin/scenarios/grpc-scenario/src/main/java/org/apache/skywalking/apm/testcase/grpc/provider/service/GreeterBlockingErrorServiceImpl.java
浏览文件 @ 97950807
......@@ -15,31 +15,17 @@
* limitations under the License.
*
*/
package org.apache.skywalking.apm.testcase.grpc.provider.service;
package org.apache.skywalking.apm.plugin.grpc.v1;
import io.grpc.CallOptions;
import io.grpc.Channel;
import io.grpc.ClientCall;
import io.grpc.ClientInterceptor;
import io.grpc.MethodDescriptor;
/**
* {@link GRPCClientInterceptor} determines the returned Interceptor based on the method type. If the method type is
* UNARY, {@link GRPCClientInterceptor} returns BlockingCallClientInterceptor, or it returns
* StreamCallClientInterceptor.
*
* @author zhang xin
*/
public class GRPCClientInterceptor implements ClientInterceptor {
import io.grpc.stub.StreamObserver;
import org.apache.skywalking.apm.testcase.grpc.proto.GreeterBlockingErrorGrpc;
import org.apache.skywalking.apm.testcase.grpc.proto.GreeterBlockingGrpc;
import org.apache.skywalking.apm.testcase.grpc.proto.HelloReply;
import org.apache.skywalking.apm.testcase.grpc.proto.HelloRequest;
public class GreeterBlockingErrorServiceImpl extends GreeterBlockingErrorGrpc.GreeterBlockingErrorImplBase {
@Override
public ClientCall interceptCall(MethodDescriptor method,
CallOptions callOptions, Channel channel) {
if (method.getType() != MethodDescriptor.MethodType.UNARY) {
return new StreamCallClientInterceptor(channel.newCall(method, callOptions), method, channel);
}
return new BlockingCallClientInterceptor(channel.newCall(method, callOptions), method, channel);
public void sayHello(HelloRequest request, StreamObserver<HelloReply> responseObserver) {
responseObserver.onError(new Exception());
}
}
test/plugin/scenarios/grpc-scenario/src/main/java/org/apache/skywalking/apm/testcase/grpc/provider/service/GreeterServiceImpl.java
浏览文件 @ 97950807
......@@ -33,6 +33,7 @@ public class GreeterServiceImpl extends GreeterGrpc.GreeterImplBase {
@Override
public StreamObserver<HelloRequest> sayHello(final StreamObserver<HelloReply> responseObserver) {
StreamObserver<HelloRequest> requestStreamObserver = new StreamObserver<HelloRequest>() {
public void onNext(HelloRequest request) {
logger.info("Receive an message from client. Message: {}", request.getName());
responseObserver.onNext(HelloReply.newBuilder().setMessage("Hi," + request.getName()).build());
......
test/plugin/scenarios/grpc-scenario/src/main/proto/GreetService.proto
浏览文件 @ 97950807
......@@ -29,6 +29,10 @@ service GreeterBlocking {
rpc SayHello (HelloRequest) returns (HelloReply) {
}
}
service GreeterBlockingError {
rpc SayHello (HelloRequest) returns (HelloReply) {
}
}
message HelloRequest {
string name = 1;
......
Markdown is supported
0% 或 .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑！
想要评论请 注册 或