diff --git a/.github/workflows/plugins-test.3.yaml b/.github/workflows/plugins-test.3.yaml index 655b1b5e2..3e9f4ffd0 100644 --- a/.github/workflows/plugins-test.3.yaml +++ b/.github/workflows/plugins-test.3.yaml @@ -100,6 +100,7 @@ jobs: - shenyu-2.4.x-scenario - jdk-threadpool-scenario - shenyu-2.4.x-dubbo-scenario + - grpc-generic-call-scenario steps: - uses: actions/checkout@v2 with: diff --git a/CHANGES.md b/CHANGES.md index 8945f8364..3605abad9 100644 --- a/CHANGES.md +++ b/CHANGES.md @@ -13,6 +13,7 @@ Release Notes. * Add an optional plugin, trace-sampler-cpu-policy-plugin, which could disable trace collecting in high CPU load. * Change the dateformat of logs to `yyyy-MM-dd HH:mm:ss.SSS`(was `yyyy-MM-dd HH:mm:ss:SSS`). * Fix NPE in elasticsearch plugin. +* Grpc plugin support trace client async generic call(without grpc stubs), support Method type: `UNARY`、`SERVER_STREAMING`. #### Documentation diff --git a/apm-sniffer/apm-sdk-plugin/grpc-1.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/grpc/v1/Constants.java b/apm-sniffer/apm-sdk-plugin/grpc-1.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/grpc/v1/Constants.java index 532dffbe8..a5e8dfac8 100644 --- a/apm-sniffer/apm-sdk-plugin/grpc-1.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/grpc/v1/Constants.java +++ b/apm-sniffer/apm-sdk-plugin/grpc-1.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/grpc/v1/Constants.java @@ -67,4 +67,8 @@ public class Constants { 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"; + + public static final String GENERIC_CALL_METHOD = "GENERIC_CALL_METHOD"; + + public static final String CLIENT_STREAM_PEER = "CLIENT_STREAM_PEER"; } \ No newline at end of file diff --git a/apm-sniffer/apm-sdk-plugin/grpc-1.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/grpc/v1/client/ClientCallImplGenericCallInterceptor.java b/apm-sniffer/apm-sdk-plugin/grpc-1.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/grpc/v1/client/ClientCallImplGenericCallInterceptor.java new file mode 100755 index 000000000..9e551275b --- /dev/null +++ b/apm-sniffer/apm-sdk-plugin/grpc-1.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/grpc/v1/client/ClientCallImplGenericCallInterceptor.java @@ -0,0 +1,224 @@ +/* + * 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.client; + +import static org.apache.skywalking.apm.plugin.grpc.v1.Constants.CLIENT; +import static org.apache.skywalking.apm.plugin.grpc.v1.Constants.CLIENT_STREAM_PEER; +import static org.apache.skywalking.apm.plugin.grpc.v1.Constants.GENERIC_CALL_METHOD; +import static org.apache.skywalking.apm.plugin.grpc.v1.Constants.RESPONSE_ON_MESSAGE_OPERATION_NAME; +import static org.apache.skywalking.apm.plugin.grpc.v1.OperationNameFormatUtil.formatOperationName; + +import java.lang.reflect.Method; +import java.util.Arrays; + +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.tag.Tags; +import org.apache.skywalking.apm.agent.core.context.trace.AbstractSpan; +import org.apache.skywalking.apm.agent.core.context.trace.ExitSpan; +import org.apache.skywalking.apm.agent.core.context.trace.SpanLayer; +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 org.apache.skywalking.apm.network.trace.component.ComponentsDefine; +import org.apache.skywalking.apm.plugin.grpc.v1.OperationNameFormatUtil; +import org.apache.skywalking.apm.util.StringUtil; + +import com.google.common.base.Strings; + +import io.grpc.Attributes; +import io.grpc.ClientCall.Listener; +import io.grpc.ForwardingClientCallListener; +import io.grpc.Metadata; +import io.grpc.MethodDescriptor; +import io.grpc.Status; +import lombok.extern.slf4j.Slf4j; + +@Slf4j +public class ClientCallImplGenericCallInterceptor + implements InstanceMethodsAroundInterceptor, InstanceConstructorInterceptor { + + @Override + public void onConstruct(EnhancedInstance objInst, Object[] allArguments) throws Throwable { + MethodDescriptor methodDescriptor = (MethodDescriptor) allArguments[0]; + objInst.setSkyWalkingDynamicField(methodDescriptor); + } + + @Override + public void beforeMethod(EnhancedInstance objInst, Method method, Object[] allArguments, Class[] argumentsTypes, + MethodInterceptResult result) throws Throwable { + String asyncCallMethod = (String) ContextManager.getRuntimeContext().get(GENERIC_CALL_METHOD); + // only trace generic call + if (StringUtil.isBlank(asyncCallMethod)) { + return; + } + ContextManager.getRuntimeContext().remove(GENERIC_CALL_METHOD); + + Listener observer = (Listener) allArguments[0]; + Metadata headers = (Metadata) allArguments[1]; + MethodDescriptor methodDescriptor = (MethodDescriptor) objInst.getSkyWalkingDynamicField(); + String serviceName = formatOperationName(methodDescriptor); + // channel.authority() In some scenes, it is not accurate. eg:Load balancing, NameResolver + // The server IP and PORT can be obtained accurately BY clientStream. + // afterMethod method will set remotePeer. + String remotePeer = "No Peer"; + String operationPrefix = OperationNameFormatUtil.formatOperationName(methodDescriptor) + CLIENT; + + ContextCarrier contextCarrier = new ContextCarrier(); + AbstractSpan span = ContextManager.createExitSpan(serviceName, contextCarrier, remotePeer); + span.setComponent(ComponentsDefine.GRPC); + span.setLayer(SpanLayer.RPC_FRAMEWORK); + span.tag(Tags.ofKey(GENERIC_CALL_METHOD), asyncCallMethod); + + CarrierItem contextItem = contextCarrier.items(); + while (contextItem.hasNext()) { + contextItem = contextItem.next(); + Metadata.Key headerKey = + Metadata.Key.of(contextItem.getHeadKey(), Metadata.ASCII_STRING_MARSHALLER); + headers.put(headerKey, contextItem.getHeadValue()); + } + ContextSnapshot snapshot = ContextManager.capture(); + span.prepareForAsync(); + ContextManager.stopSpan(span); + objInst.setSkyWalkingDynamicField(span); + + allArguments[0] = new TracingClientCallListener<>(observer, methodDescriptor, operationPrefix, snapshot, span); + } + + @Override + public Object afterMethod(EnhancedInstance objInst, Method method, Object[] allArguments, Class[] argumentsTypes, + Object ret) throws Throwable { + if (objInst.getSkyWalkingDynamicField() == null + || !(objInst.getSkyWalkingDynamicField() instanceof AbstractSpan)) { + return ret; + } + AbstractSpan span = (AbstractSpan) objInst.getSkyWalkingDynamicField(); + // Scenario of specifying IP + port + String remotePeer = (String) ContextManager.getRuntimeContext().get(CLIENT_STREAM_PEER); + ContextManager.getRuntimeContext().remove(CLIENT_STREAM_PEER); + span.setPeer(remotePeer); + Arrays.stream(objInst.getClass().getDeclaredMethods()).filter(m -> m.getName().equals("getAttributes")) + .findFirst() + .ifPresent(m -> { + try { + m.setAccessible(true); + Attributes attributes = (Attributes) m.invoke(objInst); + attributes.keys().stream() + .filter(k -> k.toString().equals("remote-addr")).findFirst() + .map(attributes::get) + .ifPresent(v -> { + String peer = v.toString(); + if (StringUtil.isNotBlank(peer)) { + if (peer.startsWith("/")) { + peer = peer.substring(1); + } + // Accurate IP acquisition.Scenario: Name Resolver , load balancing + span.setPeer(peer); + } + }); + } catch (Exception e) { + // ignore + } + }); + return ret; + } + + @Override + public void handleMethodException(EnhancedInstance objInst, Method method, Object[] allArguments, + Class[] argumentsTypes, Throwable t) { + AbstractSpan span = (AbstractSpan) objInst.getSkyWalkingDynamicField(); + if (span != null) { + span.errorOccurred().log(t); + } + } + + static class TracingClientCallListener + extends ForwardingClientCallListener.SimpleForwardingClientCallListener { + + private final ContextSnapshot contextSnapshot; + + private final MethodDescriptor methodDescriptor; + + private final String operationPrefix; + + private final AbstractSpan asyncSpan; + + TracingClientCallListener(Listener delegate, MethodDescriptor methodDescriptor, + String operationPrefix, ContextSnapshot contextSnapshot, AbstractSpan asyncSpan) { + super(delegate); + this.methodDescriptor = methodDescriptor; + this.operationPrefix = operationPrefix; + this.contextSnapshot = contextSnapshot; + this.asyncSpan = asyncSpan; + } + + @Override + public void onMessage(RESPONSE message) { + if (methodDescriptor.getType().serverSendsOneMessage()) { + super.onMessage(message); + } else { + // tracing SERVER_STREAMING + final AbstractSpan span = ContextManager + .createLocalSpan(operationPrefix + RESPONSE_ON_MESSAGE_OPERATION_NAME); + span.setComponent(ComponentsDefine.GRPC); + span.setLayer(SpanLayer.RPC_FRAMEWORK); + ContextManager.continued(contextSnapshot); + try { + delegate().onMessage(message); + } catch (Throwable t) { + span.log(t); + throw t; + } finally { + ContextManager.stopSpan(span); + } + } + } + + @Override + public void onClose(Status status, Metadata trailers) { + if (!status.isOk()) { + asyncSpan.log(status.asRuntimeException()); + } + Tags.RPC_RESPONSE_STATUS_CODE.set(asyncSpan, status.getCode().name()); + try { + delegate().onClose(status, trailers); + } catch (Throwable t) { + asyncSpan.log(t); + throw t; + } finally { + // finish async exitSpan + if (asyncSpan instanceof ExitSpan + && ContextManager.getRuntimeContext().get(CLIENT_STREAM_PEER) != null) { + // why need this? because first grpc call will create PendingStream(Unable to get IP) + // Delayed create NettyClientStream. + // In this case, the constructor of NettyClientStream and the onClose are executed in the same thread + ExitSpan exitSpan = (ExitSpan) asyncSpan; + if (Strings.isNullOrEmpty(exitSpan.getPeer())) { + asyncSpan.setPeer((String) ContextManager.getRuntimeContext().get(CLIENT_STREAM_PEER)); + ContextManager.getRuntimeContext().remove(CLIENT_STREAM_PEER); + } + } + asyncSpan.asyncFinish(); + } + } + } +} diff --git a/apm-sniffer/apm-sdk-plugin/grpc-1.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/grpc/v1/client/ClientCallsGenericCallInterceptor.java b/apm-sniffer/apm-sdk-plugin/grpc-1.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/grpc/v1/client/ClientCallsGenericCallInterceptor.java new file mode 100755 index 000000000..420c962c5 --- /dev/null +++ b/apm-sniffer/apm-sdk-plugin/grpc-1.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/grpc/v1/client/ClientCallsGenericCallInterceptor.java @@ -0,0 +1,55 @@ +/* + * 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.client; + +import static org.apache.skywalking.apm.plugin.grpc.v1.Constants.GENERIC_CALL_METHOD; + +import java.lang.reflect.Method; + +import org.apache.skywalking.apm.agent.core.context.ContextManager; +import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.MethodInterceptResult; +import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.StaticMethodsAroundInterceptor; + +/** + * Trace calling services without grpc stubs. + * This type of request is characterized by the parameter DynamicMessage + */ +public class ClientCallsGenericCallInterceptor implements StaticMethodsAroundInterceptor { + + @Override + public void beforeMethod(Class clazz, Method method, Object[] allArguments, Class[] parameterTypes, + MethodInterceptResult result) { + // only trace async generic call. Determine if the request parameter is a DynamicMessage + if (allArguments[1] instanceof com.google.protobuf.DynamicMessage) { + ContextManager.getRuntimeContext().put(GENERIC_CALL_METHOD, method.getName()); + } + } + + @Override + public Object afterMethod(Class clazz, Method method, Object[] allArguments, Class[] parameterTypes, + Object ret) { + return ret; + } + + @Override + public void handleMethodException(Class clazz, Method method, Object[] allArguments, Class[] parameterTypes, + Throwable t) { + } + +} diff --git a/apm-sniffer/apm-sdk-plugin/grpc-1.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/grpc/v1/client/NettyClientStreamInterceptor.java b/apm-sniffer/apm-sdk-plugin/grpc-1.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/grpc/v1/client/NettyClientStreamInterceptor.java new file mode 100644 index 000000000..17c4ee097 --- /dev/null +++ b/apm-sniffer/apm-sdk-plugin/grpc-1.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/grpc/v1/client/NettyClientStreamInterceptor.java @@ -0,0 +1,44 @@ +/* + * 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.client; + +import static org.apache.skywalking.apm.plugin.grpc.v1.Constants.CLIENT_STREAM_PEER; + +import org.apache.skywalking.apm.agent.core.context.ContextManager; +import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.EnhancedInstance; +import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.InstanceConstructorInterceptor; + +import lombok.extern.slf4j.Slf4j; + +/** + * Intercept constructor to obtain server IP. + */ +@Slf4j +public class NettyClientStreamInterceptor implements InstanceConstructorInterceptor { + + @Override + public void onConstruct(EnhancedInstance objInst, Object[] allArguments) throws Throwable { + String authorityClass = allArguments[4].getClass().getName(); + if ("io.netty.util.AsciiString".equals(authorityClass) + || "io.grpc.netty.shaded.io.netty.util.AsciiString".equals(authorityClass)) { + ContextManager.getRuntimeContext().put(CLIENT_STREAM_PEER, allArguments[4].toString()); + } + } + +} diff --git a/apm-sniffer/apm-sdk-plugin/grpc-1.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/grpc/v1/define/ClientCallImplInstrumentation.java b/apm-sniffer/apm-sdk-plugin/grpc-1.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/grpc/v1/define/ClientCallImplInstrumentation.java new file mode 100755 index 000000000..453748931 --- /dev/null +++ b/apm-sniffer/apm-sdk-plugin/grpc-1.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/grpc/v1/define/ClientCallImplInstrumentation.java @@ -0,0 +1,89 @@ +/* + * 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.define; + +import static net.bytebuddy.matcher.ElementMatchers.any; +import static net.bytebuddy.matcher.ElementMatchers.named; +import static org.apache.skywalking.apm.agent.core.plugin.match.NameMatch.byName; + +import org.apache.skywalking.apm.agent.core.plugin.interceptor.ConstructorInterceptPoint; +import org.apache.skywalking.apm.agent.core.plugin.interceptor.InstanceMethodsInterceptPoint; +import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.ClassInstanceMethodsEnhancePluginDefine; +import org.apache.skywalking.apm.agent.core.plugin.match.ClassMatch; + +import net.bytebuddy.description.method.MethodDescription; +import net.bytebuddy.matcher.ElementMatcher; + +/** + * create exitSpan in io.grpc.internal.ClientCallImpl#start method. + */ +public class ClientCallImplInstrumentation extends ClassInstanceMethodsEnhancePluginDefine { + + private static final String ENHANCE_CLASS = "io.grpc.internal.ClientCallImpl"; + + private static final String ENHANCE_METHOD = "start"; + + private static final String INTERCEPTOR_CLASS = + "org.apache.skywalking.apm.plugin.grpc.v1.client.ClientCallImplGenericCallInterceptor"; + + @Override + public ConstructorInterceptPoint[] getConstructorsInterceptPoints() { + return new ConstructorInterceptPoint[] { + new ConstructorInterceptPoint() { + @Override + public ElementMatcher getConstructorMatcher() { + return any(); + } + + @Override + public String getConstructorInterceptor() { + return INTERCEPTOR_CLASS; + } + } + }; + } + + @Override + public InstanceMethodsInterceptPoint[] getInstanceMethodsInterceptPoints() { + return new InstanceMethodsInterceptPoint[] { + new InstanceMethodsInterceptPoint() { + @Override + public ElementMatcher getMethodsMatcher() { + return named(ENHANCE_METHOD); + } + + @Override + public String getMethodsInterceptor() { + return INTERCEPTOR_CLASS; + } + + @Override + public boolean isOverrideArgs() { + return true; + } + } + }; + } + + @Override + protected ClassMatch enhanceClass() { + return byName(ENHANCE_CLASS); + } + +} diff --git a/apm-sniffer/apm-sdk-plugin/grpc-1.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/grpc/v1/define/ClientCallsGenericCallInstrumentation.java b/apm-sniffer/apm-sdk-plugin/grpc-1.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/grpc/v1/define/ClientCallsGenericCallInstrumentation.java new file mode 100755 index 000000000..b1a73cbed --- /dev/null +++ b/apm-sniffer/apm-sdk-plugin/grpc-1.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/grpc/v1/define/ClientCallsGenericCallInstrumentation.java @@ -0,0 +1,100 @@ +/* + * 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.define; + +import static net.bytebuddy.matcher.ElementMatchers.named; +import static org.apache.skywalking.apm.agent.core.plugin.match.NameMatch.byName; + +import org.apache.skywalking.apm.agent.core.plugin.interceptor.StaticMethodsInterceptPoint; +import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.ClassStaticMethodsEnhancePluginDefine; +import org.apache.skywalking.apm.agent.core.plugin.match.ClassMatch; + +import net.bytebuddy.description.method.MethodDescription; +import net.bytebuddy.matcher.ElementMatcher; + +/** + * trace generic call(Calling services without stubs). + */ +public class ClientCallsGenericCallInstrumentation 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.client.ClientCallsGenericCallInterceptor"; + + @Override + public StaticMethodsInterceptPoint[] getStaticMethodsInterceptPoints() { + return new StaticMethodsInterceptPoint[] { + new StaticMethodsInterceptPoint() { + @Override + public ElementMatcher getMethodsMatcher() { + return named("asyncUnaryCall"); + } + + @Override + public String getMethodsInterceptor() { + return INTERCEPTOR_CLASS; + } + + @Override + public boolean isOverrideArgs() { + return false; + } + }, + new StaticMethodsInterceptPoint() { + @Override + public ElementMatcher getMethodsMatcher() { + return named("asyncServerStreamingCall"); + } + + @Override + public String getMethodsInterceptor() { + return INTERCEPTOR_CLASS; + } + + @Override + public boolean isOverrideArgs() { + return false; + } + }, + new StaticMethodsInterceptPoint() { + @Override + public ElementMatcher getMethodsMatcher() { + return named("futureUnaryCall"); + } + + @Override + public String getMethodsInterceptor() { + return INTERCEPTOR_CLASS; + } + + @Override + public boolean isOverrideArgs() { + return false; + } + } + }; + } + + @Override + protected ClassMatch enhanceClass() { + return byName(ENHANCE_CLASS); + } + +} diff --git a/apm-sniffer/apm-sdk-plugin/grpc-1.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/grpc/v1/define/NettyClientStreamInstrumentation.java b/apm-sniffer/apm-sdk-plugin/grpc-1.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/grpc/v1/define/NettyClientStreamInstrumentation.java new file mode 100644 index 000000000..dfe74285c --- /dev/null +++ b/apm-sniffer/apm-sdk-plugin/grpc-1.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/grpc/v1/define/NettyClientStreamInstrumentation.java @@ -0,0 +1,71 @@ +/* + * 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.define; + +import static net.bytebuddy.matcher.ElementMatchers.any; +import static org.apache.skywalking.apm.agent.core.plugin.match.NameMatch.byName; + +import org.apache.skywalking.apm.agent.core.plugin.interceptor.ConstructorInterceptPoint; +import org.apache.skywalking.apm.agent.core.plugin.interceptor.InstanceMethodsInterceptPoint; +import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.ClassInstanceMethodsEnhancePluginDefine; +import org.apache.skywalking.apm.agent.core.plugin.match.ClassMatch; + +import net.bytebuddy.description.method.MethodDescription; +import net.bytebuddy.matcher.ElementMatcher; + +/** + * Intercept constructor to obtain server IP. + * grpc Low version(eg:1.6.0): use io.grpc.netty.NettyClientStream. + * grpc High version(eg:1.33.1): use io.grpc.netty.shaded.io.grpc.netty.NettyClientStream. + */ +public class NettyClientStreamInstrumentation extends ClassInstanceMethodsEnhancePluginDefine { + + private static final String ENHANCE_CLASS = "io.grpc.netty.NettyClientStream"; + + private static final String INTERCEPTOR_CLASS = + "org.apache.skywalking.apm.plugin.grpc.v1.client.NettyClientStreamInterceptor"; + + @Override + protected ClassMatch enhanceClass() { + return byName(ENHANCE_CLASS); + } + + @Override + public ConstructorInterceptPoint[] getConstructorsInterceptPoints() { + return new ConstructorInterceptPoint[] { + new ConstructorInterceptPoint() { + @Override + public ElementMatcher getConstructorMatcher() { + // only one constructor + return any(); + } + + @Override + public String getConstructorInterceptor() { + return INTERCEPTOR_CLASS; + } + } + }; + } + + @Override + public InstanceMethodsInterceptPoint[] getInstanceMethodsInterceptPoints() { + return new InstanceMethodsInterceptPoint[0]; + } +} diff --git a/apm-sniffer/apm-sdk-plugin/grpc-1.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/grpc/v1/define/ShadedNettyClientStreamInstrumentation.java b/apm-sniffer/apm-sdk-plugin/grpc-1.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/grpc/v1/define/ShadedNettyClientStreamInstrumentation.java new file mode 100644 index 000000000..eb13db910 --- /dev/null +++ b/apm-sniffer/apm-sdk-plugin/grpc-1.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/grpc/v1/define/ShadedNettyClientStreamInstrumentation.java @@ -0,0 +1,71 @@ +/* + * 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.define; + +import static net.bytebuddy.matcher.ElementMatchers.any; +import static org.apache.skywalking.apm.agent.core.plugin.match.NameMatch.byName; + +import org.apache.skywalking.apm.agent.core.plugin.interceptor.ConstructorInterceptPoint; +import org.apache.skywalking.apm.agent.core.plugin.interceptor.InstanceMethodsInterceptPoint; +import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.ClassInstanceMethodsEnhancePluginDefine; +import org.apache.skywalking.apm.agent.core.plugin.match.ClassMatch; + +import net.bytebuddy.description.method.MethodDescription; +import net.bytebuddy.matcher.ElementMatcher; + +/** + * Intercept constructor to obtain server IP. + * grpc Low version(eg:1.6.0): use io.grpc.netty.NettyClientStream. + * grpc High version(eg:1.33.1): use io.grpc.netty.shaded.io.grpc.netty.NettyClientStream. + */ +public class ShadedNettyClientStreamInstrumentation extends ClassInstanceMethodsEnhancePluginDefine { + + private static final String ENHANCE_CLASS = "io.grpc.netty.shaded.io.grpc.netty.NettyClientStream"; + + private static final String INTERCEPTOR_CLASS = + "org.apache.skywalking.apm.plugin.grpc.v1.client.NettyClientStreamInterceptor"; + + @Override + protected ClassMatch enhanceClass() { + return byName(ENHANCE_CLASS); + } + + @Override + public ConstructorInterceptPoint[] getConstructorsInterceptPoints() { + return new ConstructorInterceptPoint[] { + new ConstructorInterceptPoint() { + @Override + public ElementMatcher getConstructorMatcher() { + // only one constructor + return any(); + } + + @Override + public String getConstructorInterceptor() { + return INTERCEPTOR_CLASS; + } + } + }; + } + + @Override + public InstanceMethodsInterceptPoint[] getInstanceMethodsInterceptPoints() { + return new InstanceMethodsInterceptPoint[0]; + } +} diff --git a/apm-sniffer/apm-sdk-plugin/grpc-1.x-plugin/src/main/resources/skywalking-plugin.def b/apm-sniffer/apm-sdk-plugin/grpc-1.x-plugin/src/main/resources/skywalking-plugin.def index 7a151a7e3..11a137f4e 100644 --- a/apm-sniffer/apm-sdk-plugin/grpc-1.x-plugin/src/main/resources/skywalking-plugin.def +++ b/apm-sniffer/apm-sdk-plugin/grpc-1.x-plugin/src/main/resources/skywalking-plugin.def @@ -16,4 +16,8 @@ grpc-1.x=org.apache.skywalking.apm.plugin.grpc.v1.define.AbstractStubInstrumentation grpc-1.x=org.apache.skywalking.apm.plugin.grpc.v1.define.AbstractServerImplBuilderInstrumentation -grpc-1.x=org.apache.skywalking.apm.plugin.grpc.v1.define.ClientCallsInstrumentation \ No newline at end of file +grpc-1.x=org.apache.skywalking.apm.plugin.grpc.v1.define.ClientCallsInstrumentation +grpc-1.x=org.apache.skywalking.apm.plugin.grpc.v1.define.ClientCallsGenericCallInstrumentation +grpc-1.x=org.apache.skywalking.apm.plugin.grpc.v1.define.ClientCallImplInstrumentation +grpc-1.x=org.apache.skywalking.apm.plugin.grpc.v1.define.NettyClientStreamInstrumentation +grpc-1.x=org.apache.skywalking.apm.plugin.grpc.v1.define.ShadedNettyClientStreamInstrumentation \ No newline at end of file diff --git a/test/plugin/scenarios/grpc-generic-call-scenario/config/expectedData.yaml b/test/plugin/scenarios/grpc-generic-call-scenario/config/expectedData.yaml new file mode 100644 index 000000000..0d214640a --- /dev/null +++ b/test/plugin/scenarios/grpc-generic-call-scenario/config/expectedData.yaml @@ -0,0 +1,105 @@ +# 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. +segmentItems: + - serviceName: grpc-provider + segmentSize: nq 0 + segments: + - segmentId: not null + spans: + - operationName: org.apache.skywalking.apm.testcase.grpc.proto.TestServiceDynamic.unary/server/Response/onClose + parentSpanId: 0 + spanId: 1 + spanLayer: RPCFramework + startTime: nq 0 + endTime: nq 0 + componentId: 23 + isError: false + spanType: Local + peer: '' + skipAnalysis: false + tags: + - {key: rpc.status_code, value: OK} + - operationName: org.apache.skywalking.apm.testcase.grpc.proto.TestServiceDynamic.unary/server/Request/onComplete + parentSpanId: -1 + spanId: 0 + spanLayer: RPCFramework + startTime: nq 0 + endTime: nq 0 + componentId: 23 + isError: false + spanType: Local + peer: '' + skipAnalysis: false + refs: + - { parentEndpoint: org.apache.skywalking.apm.testcase.grpc.proto.TestServiceDynamic.unary, + networkAddress: '', refType: CrossThread, + parentSpanId: 0, parentTraceSegmentId: not null, + parentServiceInstance: not null, parentService: grpc-provider, + traceId: not null } + - segmentId: not null + spans: + - operationName: org.apache.skywalking.apm.testcase.grpc.proto.TestServiceDynamic.unary + parentSpanId: -1 + spanId: 0 + spanLayer: RPCFramework + startTime: nq 0 + endTime: nq 0 + componentId: 23 + isError: false + spanType: Entry + peer: '' + skipAnalysis: false + refs: + - { parentEndpoint: 'GET:/case/generic-call', networkAddress: not null, + refType: CrossProcess, parentSpanId: 1, parentTraceSegmentId: not null, + parentServiceInstance: not null, parentService: grpc-consumer, + traceId: not null } + + - serviceName: grpc-consumer + segmentSize: nq 0 + segments: + - segmentId: not null + spans: + - operationName: org.apache.skywalking.apm.testcase.grpc.proto.TestServiceDynamic.unary + parentSpanId: 0 + spanId: 1 + spanLayer: RPCFramework + startTime: nq 0 + endTime: nq 0 + componentId: 23 + isError: false + spanType: Exit + peer: 'localhost:18080' + skipAnalysis: false + tags: + - {key: GENERIC_CALL_METHOD, value: asyncUnaryCall} + - {key: rpc.status_code, value: OK} + - operationName: GET:/case/generic-call + parentSpanId: -1 + spanId: 0 + spanLayer: Http + startTime: nq 0 + endTime: nq 0 + componentId: 14 + isError: false + spanType: Entry + peer: '' + skipAnalysis: false + tags: + - {key: url, value: 'http://localhost:8888/case/generic-call'} + - {key: http.method, value: GET} + +meterItems: [] \ No newline at end of file diff --git a/test/plugin/scenarios/grpc-generic-call-scenario/configuration.yml b/test/plugin/scenarios/grpc-generic-call-scenario/configuration.yml new file mode 100644 index 000000000..9b232529b --- /dev/null +++ b/test/plugin/scenarios/grpc-generic-call-scenario/configuration.yml @@ -0,0 +1,22 @@ +# 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. + +type: jvm +entryService: http://localhost:8888/case/generic-call +healthCheck: http://localhost:8888/case/healthCheck +startScript: ./bin/startup.sh +environment: +dependencies: diff --git a/test/plugin/scenarios/grpc-generic-call-scenario/grpc-consumer/pom.xml b/test/plugin/scenarios/grpc-generic-call-scenario/grpc-consumer/pom.xml new file mode 100644 index 000000000..4150dc06f --- /dev/null +++ b/test/plugin/scenarios/grpc-generic-call-scenario/grpc-consumer/pom.xml @@ -0,0 +1,79 @@ + + + + + + grpc-generic-call-scenario + org.apache.skywalking + 1.0.0 + + 4.0.0 + + grpc-consumer + + + 8 + 8 + + + + + + io.grpc + grpc-all + + + + org.apache.commons + commons-lang3 + + + + + com.alibaba + fastjson + + + + org.springframework.boot + spring-boot-starter-web + + + + + grpc-consumer + + + org.springframework.boot + spring-boot-maven-plugin + ${spring.boot.version} + + + + repackage + + + + + + + + \ No newline at end of file diff --git a/test/plugin/scenarios/grpc-generic-call-scenario/grpc-consumer/src/main/java/test/apache/skywalking/apm/testcase/grpc/generic/call/consumer/GrpcConsumerApplication.java b/test/plugin/scenarios/grpc-generic-call-scenario/grpc-consumer/src/main/java/test/apache/skywalking/apm/testcase/grpc/generic/call/consumer/GrpcConsumerApplication.java new file mode 100644 index 000000000..98f8fc7a0 --- /dev/null +++ b/test/plugin/scenarios/grpc-generic-call-scenario/grpc-consumer/src/main/java/test/apache/skywalking/apm/testcase/grpc/generic/call/consumer/GrpcConsumerApplication.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 test.apache.skywalking.apm.testcase.grpc.generic.call.consumer; + +import org.springframework.boot.SpringApplication; +import org.springframework.boot.autoconfigure.SpringBootApplication; + +@SpringBootApplication +public class GrpcConsumerApplication { + + public static void main(final String[] args) { + SpringApplication.run(GrpcConsumerApplication.class, args); + } +} diff --git a/test/plugin/scenarios/grpc-generic-call-scenario/grpc-consumer/src/main/java/test/apache/skywalking/apm/testcase/grpc/generic/call/consumer/client/DynamicJsonMessage.java b/test/plugin/scenarios/grpc-generic-call-scenario/grpc-consumer/src/main/java/test/apache/skywalking/apm/testcase/grpc/generic/call/consumer/client/DynamicJsonMessage.java new file mode 100755 index 000000000..de7149bbc --- /dev/null +++ b/test/plugin/scenarios/grpc-generic-call-scenario/grpc-consumer/src/main/java/test/apache/skywalking/apm/testcase/grpc/generic/call/consumer/client/DynamicJsonMessage.java @@ -0,0 +1,134 @@ +/* + * 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 test.apache.skywalking.apm.testcase.grpc.generic.call.consumer.client; + +import java.io.IOException; +import java.io.InputStream; +import java.util.Map; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import com.google.protobuf.DescriptorProtos; +import com.google.protobuf.Descriptors; +import com.google.protobuf.DynamicMessage; +import com.google.protobuf.ExtensionRegistryLite; + +import io.grpc.MethodDescriptor; + +public class DynamicJsonMessage { + public static final String DYNAMIC_SERVICE_SUFFIX = "Dynamic"; + + public static final String DYNAMIC_MESSAGE_NAME = "DynamicJsonMessage"; + + public static final String DYNAMIC_MESSAGE_DATA_FILED = "data"; + + private static final Logger LOG = LoggerFactory.getLogger(DynamicJsonMessage.class); + + private static Descriptors.Descriptor buildJsonMarshallerDescriptor() { + + DescriptorProtos.DescriptorProto.Builder jsonMarshaller = DescriptorProtos.DescriptorProto.newBuilder(); + jsonMarshaller.setName(DYNAMIC_MESSAGE_NAME); + jsonMarshaller.addFieldBuilder() + .setName(DYNAMIC_MESSAGE_DATA_FILED) + .setNumber(1) + .setType(DescriptorProtos.FieldDescriptorProto.Type.TYPE_STRING); + + DescriptorProtos.FileDescriptorProto.Builder fileDescriptorProtoBuilder = + DescriptorProtos.FileDescriptorProto.newBuilder(); + fileDescriptorProtoBuilder.addMessageType(jsonMarshaller); + + DescriptorProtos.FileDescriptorProto fileDescriptorProto = fileDescriptorProtoBuilder.build(); + try { + Descriptors.FileDescriptor fileDescriptor = Descriptors.FileDescriptor + .buildFrom(fileDescriptorProto, new Descriptors.FileDescriptor[0]); + return fileDescriptor.findMessageTypeByName(DYNAMIC_MESSAGE_NAME); + } catch (Exception e) { + LOG.error("buildJsonMarshallerDescriptor error: {}", e.getMessage()); + throw new RuntimeException("buildJsonMarshallerDescriptor error", e); + } + } + + public static DynamicMessage buildJsonDynamicMessage(String jsonParam) { + // build Descriptor and set request param + Descriptors.Descriptor jsonDescriptor = buildJsonMarshallerDescriptor(); + DynamicMessage.Builder jsonDynamicMessage = DynamicMessage.newBuilder(jsonDescriptor); + jsonDynamicMessage.setField(jsonDescriptor.findFieldByName(DYNAMIC_MESSAGE_DATA_FILED), + jsonParam); + return jsonDynamicMessage.build(); + } + + public static DynamicMessage buildJsonDynamicMessage() { + Descriptors.Descriptor jsonDescriptor = buildJsonMarshallerDescriptor(); + DynamicMessage.Builder jsonDynamicMessage = DynamicMessage.newBuilder(jsonDescriptor); + return jsonDynamicMessage.build(); + } + + public static String getDataFromDynamicMessage(DynamicMessage message) { + for (Map.Entry entry : message.getAllFields().entrySet()) { + Descriptors.FieldDescriptor key = entry.getKey(); + Object value = entry.getValue(); + String fullName = key.getFullName(); + String jsonMessageFullName = DYNAMIC_MESSAGE_NAME + "." + DYNAMIC_MESSAGE_DATA_FILED; + if (jsonMessageFullName.equals(fullName)) { + return (String) value; + } + } + return ""; + } + + public static MethodDescriptor createJsonMarshallerMethodDescriptor( + String serviceName, String methodName, MethodDescriptor.MethodType methodType, + DynamicMessage request, DynamicMessage response) { + + return MethodDescriptor.newBuilder() + .setType(methodType) + .setFullMethodName( + MethodDescriptor.generateFullMethodName(serviceName + DYNAMIC_SERVICE_SUFFIX, + methodName)) + .setRequestMarshaller(new DynamicMessageMarshaller(request.getDescriptorForType())) + .setResponseMarshaller(new DynamicMessageMarshaller(response.getDescriptorForType())) + .build(); + } + + private static class DynamicMessageMarshaller implements MethodDescriptor.Marshaller { + + private Descriptors.Descriptor messageDescriptor; + + private DynamicMessageMarshaller(Descriptors.Descriptor messageDescriptor) { + this.messageDescriptor = messageDescriptor; + } + + @Override + public DynamicMessage parse(InputStream inputStream) { + try { + return DynamicMessage.newBuilder(messageDescriptor) + .mergeFrom(inputStream, ExtensionRegistryLite.getEmptyRegistry()) + .build(); + } catch (IOException e) { + throw new RuntimeException("parse inputStream error", e); + } + } + + @Override + public InputStream stream(DynamicMessage abstractMessage) { + return abstractMessage.toByteString().newInput(); + } + } +} diff --git a/test/plugin/scenarios/grpc-generic-call-scenario/grpc-consumer/src/main/java/test/apache/skywalking/apm/testcase/grpc/generic/call/consumer/client/GrpcGenericCallClient.java b/test/plugin/scenarios/grpc-generic-call-scenario/grpc-consumer/src/main/java/test/apache/skywalking/apm/testcase/grpc/generic/call/consumer/client/GrpcGenericCallClient.java new file mode 100755 index 000000000..2a429334c --- /dev/null +++ b/test/plugin/scenarios/grpc-generic-call-scenario/grpc-consumer/src/main/java/test/apache/skywalking/apm/testcase/grpc/generic/call/consumer/client/GrpcGenericCallClient.java @@ -0,0 +1,82 @@ +/* + * 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 test.apache.skywalking.apm.testcase.grpc.generic.call.consumer.client; + +import static io.grpc.stub.ClientCalls.asyncServerStreamingCall; +import static io.grpc.stub.ClientCalls.asyncUnaryCall; + +import java.util.List; +import java.util.Map; +import java.util.concurrent.TimeUnit; +import java.util.stream.Collectors; + +import org.springframework.beans.factory.InitializingBean; +import org.springframework.beans.factory.annotation.Value; +import org.springframework.stereotype.Component; + +import com.alibaba.fastjson.JSON; +import com.google.protobuf.DynamicMessage; + +import io.grpc.CallOptions; +import io.grpc.ClientCall; +import io.grpc.ManagedChannel; +import io.grpc.ManagedChannelBuilder; +import io.grpc.MethodDescriptor; + +@Component +public class GrpcGenericCallClient implements InitializingBean { + + @Value("${grpc.port}") + private int grpcPort; + + private ManagedChannel channel; + + @Override + public void afterPropertiesSet() throws Exception { + channel = ManagedChannelBuilder.forAddress("localhost", grpcPort).usePlaintext(true).build(); + } + + /** + * generic call(without grpc stubs and proto) + */ + public Object genericCall(String serviceName, String methodName, MethodDescriptor.MethodType methodType, + List> paramsList) throws Exception { + // param to dynamicMessage + List jsonRequestList = paramsList.stream() + .map(params -> DynamicJsonMessage.buildJsonDynamicMessage(JSON.toJSONString(params))).collect(Collectors.toList()); + MethodDescriptor jsonMarshallerMethodDescriptor = + DynamicJsonMessage.createJsonMarshallerMethodDescriptor(serviceName, methodName, methodType, + jsonRequestList.get(0), DynamicJsonMessage.buildJsonDynamicMessage()); + SimpleStreamObserver simpleStreamObserver = new SimpleStreamObserver<>(); + ClientCall call = channel.newCall(jsonMarshallerMethodDescriptor, + CallOptions.DEFAULT.withDeadlineAfter(10000, TimeUnit.MILLISECONDS)); + + switch (methodType) { + case UNARY: + asyncUnaryCall(call, jsonRequestList.get(0), simpleStreamObserver); + return simpleStreamObserver.syncGetResponseDataList(); + case SERVER_STREAMING: + asyncServerStreamingCall(call, jsonRequestList.get(0), simpleStreamObserver); + return simpleStreamObserver.syncGetResponseDataList(); + default: + return null; + } + } + +} diff --git a/test/plugin/scenarios/grpc-generic-call-scenario/grpc-consumer/src/main/java/test/apache/skywalking/apm/testcase/grpc/generic/call/consumer/client/SimpleStreamObserver.java b/test/plugin/scenarios/grpc-generic-call-scenario/grpc-consumer/src/main/java/test/apache/skywalking/apm/testcase/grpc/generic/call/consumer/client/SimpleStreamObserver.java new file mode 100755 index 000000000..dd446e4ca --- /dev/null +++ b/test/plugin/scenarios/grpc-generic-call-scenario/grpc-consumer/src/main/java/test/apache/skywalking/apm/testcase/grpc/generic/call/consumer/client/SimpleStreamObserver.java @@ -0,0 +1,74 @@ +/* + * 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 test.apache.skywalking.apm.testcase.grpc.generic.call.consumer.client; + +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.CompletableFuture; +import java.util.stream.Collectors; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import com.alibaba.fastjson.JSON; +import com.google.protobuf.DynamicMessage; +import com.google.protobuf.Message; + +import io.grpc.stub.StreamObserver; + +public class SimpleStreamObserver implements StreamObserver { + + private static final Logger LOG = LoggerFactory.getLogger(SimpleStreamObserver.class); + + private final List responseDataList = new ArrayList<>(); + + private final CompletableFuture completableFuture = new CompletableFuture<>(); + + @Override + public void onNext(T value) { + try { + // dynamicMessage to json + String respData = DynamicJsonMessage.getDataFromDynamicMessage((DynamicMessage) value); + responseDataList.add(respData); + } catch (Exception e) { + LOG.error("parse error", e); + } + } + + @Override + public void onError(Throwable t) { + LOG.error("SimpleStreamObserver onError", t); + completableFuture.completeExceptionally(t); + } + + @Override + public void onCompleted() { + LOG.info("SimpleStreamObserver onCompleted"); + completableFuture.complete(null); + } + + public List syncGetResponseDataList() throws Exception { + completableFuture.get(); + return responseDataList.stream().map(JSON::parseObject).collect(Collectors.toList()); + } + + public CompletableFuture getCompletableFuture() { + return completableFuture; + } +} diff --git a/test/plugin/scenarios/grpc-generic-call-scenario/grpc-consumer/src/main/java/test/apache/skywalking/apm/testcase/grpc/generic/call/consumer/controller/CaseController.java b/test/plugin/scenarios/grpc-generic-call-scenario/grpc-consumer/src/main/java/test/apache/skywalking/apm/testcase/grpc/generic/call/consumer/controller/CaseController.java new file mode 100755 index 000000000..895e56800 --- /dev/null +++ b/test/plugin/scenarios/grpc-generic-call-scenario/grpc-consumer/src/main/java/test/apache/skywalking/apm/testcase/grpc/generic/call/consumer/controller/CaseController.java @@ -0,0 +1,56 @@ +/* + * 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 test.apache.skywalking.apm.testcase.grpc.generic.call.consumer.controller; + +import java.util.Map; + +import test.apache.skywalking.apm.testcase.grpc.generic.call.consumer.client.GrpcGenericCallClient; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.web.bind.annotation.RequestMapping; +import org.springframework.web.bind.annotation.ResponseBody; +import org.springframework.web.bind.annotation.RestController; + +import com.google.common.collect.ImmutableMap; +import com.google.common.collect.Lists; + +import io.grpc.MethodDescriptor.MethodType; + +@RestController +@RequestMapping("/case") +public class CaseController { + + @Autowired + private GrpcGenericCallClient grpcGenericCallClient; + + @RequestMapping("/healthCheck") + @ResponseBody + public String healthCheck() { + return "OK"; + } + + @RequestMapping("/generic-call") + @ResponseBody + public Object genericCall() throws Exception { + Thread.sleep(1000); + Map params = ImmutableMap.of("data", "hello unary"); + return grpcGenericCallClient.genericCall("org.apache.skywalking.apm.testcase.grpc.proto.TestService", + "unary", MethodType.UNARY, Lists.newArrayList(params)); + } + +} diff --git a/test/plugin/scenarios/grpc-generic-call-scenario/grpc-consumer/src/main/resources/application.yml b/test/plugin/scenarios/grpc-generic-call-scenario/grpc-consumer/src/main/resources/application.yml new file mode 100644 index 000000000..d7f4f635d --- /dev/null +++ b/test/plugin/scenarios/grpc-generic-call-scenario/grpc-consumer/src/main/resources/application.yml @@ -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. + +server: + port: 8888 +grpc: + port: 18080 diff --git a/test/plugin/scenarios/grpc-generic-call-scenario/grpc-dist/bin/startup.sh b/test/plugin/scenarios/grpc-generic-call-scenario/grpc-dist/bin/startup.sh new file mode 100644 index 000000000..a03ec2fdf --- /dev/null +++ b/test/plugin/scenarios/grpc-generic-call-scenario/grpc-dist/bin/startup.sh @@ -0,0 +1,27 @@ +#!/bin/bash +# +# 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. + +home="$(cd "$(dirname $0)"; pwd)" + +java -jar ${agent_opts} "-Dskywalking.agent.service_name=grpc-provider" ${home}/../libs/grpc-provider.jar & +sleep 2 + +java -jar ${agent_opts} "-Dskywalking.agent.service_name=grpc-consumer" ${home}/../libs/grpc-consumer.jar & +sleep 2 + + diff --git a/test/plugin/scenarios/grpc-generic-call-scenario/grpc-dist/pom.xml b/test/plugin/scenarios/grpc-generic-call-scenario/grpc-dist/pom.xml new file mode 100644 index 000000000..7574e8bf9 --- /dev/null +++ b/test/plugin/scenarios/grpc-generic-call-scenario/grpc-dist/pom.xml @@ -0,0 +1,59 @@ + + + + + grpc-generic-call-scenario + org.apache.skywalking + 1.0.0 + + 4.0.0 + + grpc-dist + + + 8 + 8 + + + + + + org.apache.maven.plugins + maven-assembly-plugin + + + assemble + package + + single + + + + src/main/assembly/assembly.xml + + ../target/ + + + + + + + \ No newline at end of file diff --git a/test/plugin/scenarios/grpc-generic-call-scenario/grpc-dist/src/main/assembly/assembly.xml b/test/plugin/scenarios/grpc-generic-call-scenario/grpc-dist/src/main/assembly/assembly.xml new file mode 100644 index 000000000..8a2ae21fa --- /dev/null +++ b/test/plugin/scenarios/grpc-generic-call-scenario/grpc-dist/src/main/assembly/assembly.xml @@ -0,0 +1,48 @@ + + + + + zip + + + + + ./bin + 0775 + + + + + + ../grpc-provider/target/grpc-provider.jar + ./libs + 0775 + + + + ../grpc-consumer/target/grpc-consumer.jar + ./libs + 0775 + + + + \ No newline at end of file diff --git a/test/plugin/scenarios/grpc-generic-call-scenario/grpc-provider/pom.xml b/test/plugin/scenarios/grpc-generic-call-scenario/grpc-provider/pom.xml new file mode 100644 index 000000000..9003c03bf --- /dev/null +++ b/test/plugin/scenarios/grpc-generic-call-scenario/grpc-provider/pom.xml @@ -0,0 +1,123 @@ + + + + + + grpc-generic-call-scenario + org.apache.skywalking + 1.0.0 + + 4.0.0 + + grpc-provider + + + 8 + 8 + 1.6.2 + + + + + + + io.grpc + grpc-all + + + + org.apache.commons + commons-lang3 + + + + + com.alibaba + fastjson + + + + org.springframework.boot + spring-boot-starter-web + + + + + + grpc-provider + + + kr.motd.maven + os-maven-plugin + ${os-maven-plugin.version} + + + initialize + + detect + + + + + + + org.springframework.boot + spring-boot-maven-plugin + ${spring.boot.version} + + + + repackage + + + + + + + org.xolstice.maven.plugins + protobuf-maven-plugin + 0.5.0 + + + com.google.protobuf:protoc:3.3.0:exe:${os.detected.classifier} + + grpc-java + io.grpc:protoc-gen-grpc-java:${test.framework.version}:exe:${os.detected.classifier} + + + + + + compile + compile-custom + + + + + + + + + \ No newline at end of file diff --git a/test/plugin/scenarios/grpc-generic-call-scenario/grpc-provider/src/main/java/org/apache/skywalking/apm/testcase/grpc/generic/call/provider/GrpcProviderApplication.java b/test/plugin/scenarios/grpc-generic-call-scenario/grpc-provider/src/main/java/org/apache/skywalking/apm/testcase/grpc/generic/call/provider/GrpcProviderApplication.java new file mode 100644 index 000000000..361dd51cf --- /dev/null +++ b/test/plugin/scenarios/grpc-generic-call-scenario/grpc-provider/src/main/java/org/apache/skywalking/apm/testcase/grpc/generic/call/provider/GrpcProviderApplication.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.skywalking.apm.testcase.grpc.generic.call.provider; + +import org.springframework.boot.SpringApplication; +import org.springframework.boot.autoconfigure.SpringBootApplication; + +@SpringBootApplication +public class GrpcProviderApplication { + + public static void main(final String[] args) { + SpringApplication.run(GrpcProviderApplication.class, args); + } +} diff --git a/test/plugin/scenarios/grpc-generic-call-scenario/grpc-provider/src/main/java/org/apache/skywalking/apm/testcase/grpc/generic/call/provider/provider/ProviderConfiguration.java b/test/plugin/scenarios/grpc-generic-call-scenario/grpc-provider/src/main/java/org/apache/skywalking/apm/testcase/grpc/generic/call/provider/provider/ProviderConfiguration.java new file mode 100755 index 000000000..680f55ff8 --- /dev/null +++ b/test/plugin/scenarios/grpc-generic-call-scenario/grpc-provider/src/main/java/org/apache/skywalking/apm/testcase/grpc/generic/call/provider/provider/ProviderConfiguration.java @@ -0,0 +1,51 @@ +/* + * 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.testcase.grpc.generic.call.provider.provider; + +import org.apache.skywalking.apm.testcase.grpc.generic.call.provider.provider.service.TestServiceImpl; +import org.apache.skywalking.apm.testcase.grpc.generic.call.provider.server.dynamic.DynamicMessageServiceTranslator; +import org.springframework.beans.factory.annotation.Value; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; + +import io.grpc.Server; +import io.grpc.ServerBuilder; +import io.grpc.ServerServiceDefinition; + +@Configuration +public class ProviderConfiguration { + + @Value("${grpc.port}") + private int grpcPort; + + @Bean(initMethod = "start", destroyMethod = "shutdown") + public Server server(TestServiceImpl testService) throws Exception { + ServerServiceDefinition serviceDefinition = testService.bindService(); + // For testing purposes, the `GRPC Server Reflection Protocol` is not used here. + // GRPC Server Reflection Protocol: https://github.com/grpc/grpc/blob/master/doc/server-reflection.md + // No dependency: io.grpc:grpc-services:${grpcVersion} + ServerServiceDefinition jsonDefinition = + DynamicMessageServiceTranslator.buildDynamicServerServiceDefinition(serviceDefinition); + return ServerBuilder.forPort(grpcPort) + .addService(serviceDefinition) + .addService(jsonDefinition) + .build(); + } + +} diff --git a/test/plugin/scenarios/grpc-generic-call-scenario/grpc-provider/src/main/java/org/apache/skywalking/apm/testcase/grpc/generic/call/provider/provider/service/TestServiceImpl.java b/test/plugin/scenarios/grpc-generic-call-scenario/grpc-provider/src/main/java/org/apache/skywalking/apm/testcase/grpc/generic/call/provider/provider/service/TestServiceImpl.java new file mode 100755 index 000000000..fe97ad44d --- /dev/null +++ b/test/plugin/scenarios/grpc-generic-call-scenario/grpc-provider/src/main/java/org/apache/skywalking/apm/testcase/grpc/generic/call/provider/provider/service/TestServiceImpl.java @@ -0,0 +1,53 @@ +/* + * 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.testcase.grpc.generic.call.provider.provider.service; + +import org.apache.skywalking.apm.testcase.grpc.proto.RequestData; +import org.apache.skywalking.apm.testcase.grpc.proto.ResponseData; +import org.apache.skywalking.apm.testcase.grpc.proto.TestServiceGrpc; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.stereotype.Service; + +import io.grpc.stub.StreamObserver; + +@Service +public class TestServiceImpl extends TestServiceGrpc.TestServiceImplBase { + + private static final Logger LOG = LoggerFactory.getLogger(TestServiceImpl.class); + + @Override + public void unary(RequestData request, StreamObserver responseObserver) { + LOG.info("server unary receive data:{}", request.getData()); + ResponseData responseData = ResponseData.newBuilder().setData("unaryFun response").build(); + responseObserver.onNext(responseData); + responseObserver.onCompleted(); + } + + @Override + public void serverStreaming(RequestData request, StreamObserver responseObserver) { + LOG.info("server serverStreaming receive data:{}", request.getData()); + for (int i = 0; i < 2; i++) { + ResponseData responseData = ResponseData.newBuilder().setData("serverStreaming response data " + i).build(); + responseObserver.onNext(responseData); + } + responseObserver.onCompleted(); + } + +} diff --git a/test/plugin/scenarios/grpc-generic-call-scenario/grpc-provider/src/main/java/org/apache/skywalking/apm/testcase/grpc/generic/call/provider/server/constant/Constants.java b/test/plugin/scenarios/grpc-generic-call-scenario/grpc-provider/src/main/java/org/apache/skywalking/apm/testcase/grpc/generic/call/provider/server/constant/Constants.java new file mode 100755 index 000000000..7bf080ca4 --- /dev/null +++ b/test/plugin/scenarios/grpc-generic-call-scenario/grpc-provider/src/main/java/org/apache/skywalking/apm/testcase/grpc/generic/call/provider/server/constant/Constants.java @@ -0,0 +1,28 @@ +/* + * 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.testcase.grpc.generic.call.provider.server.constant; + +public class Constants { + + public static final String DYNAMIC_SERVICE_SUFFIX = "Dynamic"; + + public static final String DYNAMIC_MESSAGE_NAME = "DynamicJsonMessage"; + + public static final String DYNAMIC_MESSAGE_DATA_FILED = "data"; +} diff --git a/test/plugin/scenarios/grpc-generic-call-scenario/grpc-provider/src/main/java/org/apache/skywalking/apm/testcase/grpc/generic/call/provider/server/dynamic/DynamicJsonMessage.java b/test/plugin/scenarios/grpc-generic-call-scenario/grpc-provider/src/main/java/org/apache/skywalking/apm/testcase/grpc/generic/call/provider/server/dynamic/DynamicJsonMessage.java new file mode 100755 index 000000000..00fe5cfa6 --- /dev/null +++ b/test/plugin/scenarios/grpc-generic-call-scenario/grpc-provider/src/main/java/org/apache/skywalking/apm/testcase/grpc/generic/call/provider/server/dynamic/DynamicJsonMessage.java @@ -0,0 +1,130 @@ +/* + * 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.testcase.grpc.generic.call.provider.server.dynamic; + +import java.io.IOException; +import java.io.InputStream; +import java.util.Map; + +import org.apache.skywalking.apm.testcase.grpc.generic.call.provider.server.constant.Constants; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import com.google.protobuf.DescriptorProtos; +import com.google.protobuf.Descriptors; +import com.google.protobuf.DynamicMessage; +import com.google.protobuf.ExtensionRegistryLite; + +import io.grpc.MethodDescriptor; + +public class DynamicJsonMessage { + + private static final Logger LOG = LoggerFactory.getLogger(DynamicJsonMessage.class); + + private static Descriptors.Descriptor buildJsonMarshallerDescriptor() { + + DescriptorProtos.DescriptorProto.Builder jsonMarshaller = DescriptorProtos.DescriptorProto.newBuilder(); + jsonMarshaller.setName(Constants.DYNAMIC_MESSAGE_NAME); + jsonMarshaller.addFieldBuilder() + .setName(Constants.DYNAMIC_MESSAGE_DATA_FILED) + .setNumber(1) + .setType(DescriptorProtos.FieldDescriptorProto.Type.TYPE_STRING); + + DescriptorProtos.FileDescriptorProto.Builder fileDescriptorProtoBuilder = + DescriptorProtos.FileDescriptorProto.newBuilder(); + fileDescriptorProtoBuilder.addMessageType(jsonMarshaller); + + DescriptorProtos.FileDescriptorProto fileDescriptorProto = fileDescriptorProtoBuilder.build(); + try { + Descriptors.FileDescriptor fileDescriptor = Descriptors.FileDescriptor + .buildFrom(fileDescriptorProto, new Descriptors.FileDescriptor[0]); + return fileDescriptor.findMessageTypeByName(Constants.DYNAMIC_MESSAGE_NAME); + } catch (Exception e) { + LOG.error("buildJsonMarshallerDescriptor error: {}", e.getMessage()); + throw new RuntimeException("buildJsonMarshallerDescriptor error", e); + } + } + + public static DynamicMessage buildJsonDynamicMessage(String jsonParam) { + // build Descriptor and set request param + Descriptors.Descriptor jsonDescriptor = buildJsonMarshallerDescriptor(); + DynamicMessage.Builder jsonDynamicMessage = DynamicMessage.newBuilder(jsonDescriptor); + jsonDynamicMessage.setField(jsonDescriptor.findFieldByName(Constants.DYNAMIC_MESSAGE_DATA_FILED), + jsonParam); + return jsonDynamicMessage.build(); + } + + public static DynamicMessage buildJsonDynamicMessage() { + Descriptors.Descriptor jsonDescriptor = buildJsonMarshallerDescriptor(); + DynamicMessage.Builder jsonDynamicMessage = DynamicMessage.newBuilder(jsonDescriptor); + return jsonDynamicMessage.build(); + } + + public static String getDataFromDynamicMessage(DynamicMessage message) { + for (Map.Entry entry : message.getAllFields().entrySet()) { + Descriptors.FieldDescriptor key = entry.getKey(); + Object value = entry.getValue(); + String fullName = key.getFullName(); + String jsonMessageFullName = Constants.DYNAMIC_MESSAGE_NAME + "." + Constants.DYNAMIC_MESSAGE_DATA_FILED; + if (jsonMessageFullName.equals(fullName)) { + return (String) value; + } + } + return ""; + } + + public static MethodDescriptor createJsonMarshallerMethodDescriptor( + String serviceName, String methodName, MethodDescriptor.MethodType methodType, + DynamicMessage request, DynamicMessage response) { + + return MethodDescriptor.newBuilder() + .setType(methodType) + .setFullMethodName( + MethodDescriptor.generateFullMethodName(serviceName + Constants.DYNAMIC_SERVICE_SUFFIX, + methodName)) + .setRequestMarshaller(new DynamicMessageMarshaller(request.getDescriptorForType())) + .setResponseMarshaller(new DynamicMessageMarshaller(response.getDescriptorForType())) + .build(); + } + + private static class DynamicMessageMarshaller implements MethodDescriptor.Marshaller { + + private Descriptors.Descriptor messageDescriptor; + + private DynamicMessageMarshaller(Descriptors.Descriptor messageDescriptor) { + this.messageDescriptor = messageDescriptor; + } + + @Override + public DynamicMessage parse(InputStream inputStream) { + try { + return DynamicMessage.newBuilder(messageDescriptor) + .mergeFrom(inputStream, ExtensionRegistryLite.getEmptyRegistry()) + .build(); + } catch (IOException e) { + throw new RuntimeException("parse inputStream error", e); + } + } + + @Override + public InputStream stream(DynamicMessage abstractMessage) { + return abstractMessage.toByteString().newInput(); + } + } +} diff --git a/test/plugin/scenarios/grpc-generic-call-scenario/grpc-provider/src/main/java/org/apache/skywalking/apm/testcase/grpc/generic/call/provider/server/dynamic/DynamicMessageServiceTranslator.java b/test/plugin/scenarios/grpc-generic-call-scenario/grpc-provider/src/main/java/org/apache/skywalking/apm/testcase/grpc/generic/call/provider/server/dynamic/DynamicMessageServiceTranslator.java new file mode 100755 index 000000000..cc4c87a27 --- /dev/null +++ b/test/plugin/scenarios/grpc-generic-call-scenario/grpc-provider/src/main/java/org/apache/skywalking/apm/testcase/grpc/generic/call/provider/server/dynamic/DynamicMessageServiceTranslator.java @@ -0,0 +1,126 @@ +/* + * 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.testcase.grpc.generic.call.provider.server.dynamic; + +import java.lang.reflect.Field; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.Objects; + +import org.apache.skywalking.apm.testcase.grpc.generic.call.provider.server.constant.Constants; + +import com.google.protobuf.DynamicMessage; + +import io.grpc.MethodDescriptor; +import io.grpc.MethodDescriptor.PrototypeMarshaller; +import io.grpc.ServerCall; +import io.grpc.ServerCallHandler; +import io.grpc.ServerMethodDefinition; +import io.grpc.ServerServiceDefinition; +import io.grpc.ServiceDescriptor; + +public class DynamicMessageServiceTranslator { + + private static final Map> REQUEST_PARAMS_CLASS_MAP = new HashMap<>(); + + public static ServerServiceDefinition buildDynamicServerServiceDefinition(ServerServiceDefinition serviceDef) + throws Exception { + MethodDescriptor.Marshaller marshaller = + io.grpc.protobuf.ProtoUtils.marshaller(DynamicJsonMessage.buildJsonDynamicMessage()); + List> wrappedMethods = new ArrayList<>(); + List> wrappedDescriptors = new ArrayList<>(); + + for (ServerMethodDefinition definition : serviceDef.getMethods()) { + MethodDescriptor.Marshaller requestMarshaller = definition.getMethodDescriptor().getRequestMarshaller(); + + String fullMethodName = definition.getMethodDescriptor().getFullMethodName(); + + String[] splitMethodName = fullMethodName.split("/"); + fullMethodName = splitMethodName[0] + Constants.DYNAMIC_SERVICE_SUFFIX + "/" + splitMethodName[1]; + if (requestMarshaller instanceof MethodDescriptor.PrototypeMarshaller) { + MethodDescriptor.PrototypeMarshaller prototypeMarshaller = + (PrototypeMarshaller) requestMarshaller; + REQUEST_PARAMS_CLASS_MAP.put(fullMethodName, prototypeMarshaller.getMessagePrototype().getClass()); + } + MethodDescriptor originalMethodDescriptor = definition.getMethodDescriptor(); + MethodDescriptor wrappedMethodDescriptor = originalMethodDescriptor + .toBuilder(marshaller, marshaller).build(); + wrappedDescriptors.add(wrappedMethodDescriptor); + wrappedMethods.add(wrapMethod(definition, wrappedMethodDescriptor)); + } + + ServiceDescriptor.Builder build = ServiceDescriptor.newBuilder( + serviceDef.getServiceDescriptor().getName() + Constants.DYNAMIC_SERVICE_SUFFIX); + for (MethodDescriptor md : wrappedDescriptors) { + Field fullMethodNameField = getField(md.getClass(), "fullMethodName"); + fullMethodNameField.setAccessible(true); + String fullMethodName = (String) fullMethodNameField.get(md); + String[] splitMethodName = fullMethodName.split("/"); + fullMethodName = splitMethodName[0] + Constants.DYNAMIC_SERVICE_SUFFIX + "/" + splitMethodName[1]; + fullMethodNameField.set(md, fullMethodName); + + String serviceName; + Field serviceNameField = getField(md.getClass(), "serviceName"); + if (Objects.nonNull(serviceNameField)) { + serviceNameField.setAccessible(true); + serviceName = (String) serviceNameField.get(md); + serviceName = serviceName + Constants.DYNAMIC_SERVICE_SUFFIX; + serviceNameField.set(md, serviceName); + } + build.addMethod(md); + } + ServerServiceDefinition.Builder serviceBuilder = ServerServiceDefinition.builder(build.build()); + + for (ServerMethodDefinition definition : wrappedMethods) { + serviceBuilder.addMethod(definition); + } + return serviceBuilder.build(); + } + + private static Field getField(final Class beanClass, final String name) throws SecurityException { + final Field[] fields = beanClass.getDeclaredFields(); + return Arrays.stream(fields).filter(field -> Objects.equals(name, field.getName())) + .findFirst().orElse(null); + } + + private static ServerMethodDefinition wrapMethod( + ServerMethodDefinition definition, + MethodDescriptor wrappedMethod) { + ServerCallHandler wrappedHandler = wrapHandler(definition.getServerCallHandler()); + return ServerMethodDefinition.create(wrappedMethod, wrappedHandler); + } + + @SuppressWarnings("unchecked") + private static ServerCallHandler wrapHandler( + ServerCallHandler originalHandler) { + return (call, headers) -> { + ServerCall unwrappedCall = new ProtoToDynamicServerCall<>((ServerCall) call); + ServerCall.Listener originalListener = originalHandler.startCall(unwrappedCall, headers); + return new DynamicToProtoServerCallListener(originalListener, unwrappedCall); + }; + } + + public static Map> getRequestParamsClassMap() { + return REQUEST_PARAMS_CLASS_MAP; + } + +} diff --git a/test/plugin/scenarios/grpc-generic-call-scenario/grpc-provider/src/main/java/org/apache/skywalking/apm/testcase/grpc/generic/call/provider/server/dynamic/DynamicToProtoServerCallListener.java b/test/plugin/scenarios/grpc-generic-call-scenario/grpc-provider/src/main/java/org/apache/skywalking/apm/testcase/grpc/generic/call/provider/server/dynamic/DynamicToProtoServerCallListener.java new file mode 100755 index 000000000..5443d4d9f --- /dev/null +++ b/test/plugin/scenarios/grpc-generic-call-scenario/grpc-provider/src/main/java/org/apache/skywalking/apm/testcase/grpc/generic/call/provider/server/dynamic/DynamicToProtoServerCallListener.java @@ -0,0 +1,81 @@ +/* + * 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.testcase.grpc.generic.call.provider.server.dynamic; + +import java.lang.reflect.InvocationTargetException; +import java.util.Objects; + +import org.apache.commons.lang3.reflect.MethodUtils; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import com.google.protobuf.DynamicMessage; +import com.google.protobuf.Message; +import com.google.protobuf.util.JsonFormat; + +import io.grpc.ForwardingServerCallListener.SimpleForwardingServerCallListener; +import io.grpc.ServerCall; +import io.grpc.ServerCall.Listener; +import io.grpc.Status; + +public class DynamicToProtoServerCallListener extends SimpleForwardingServerCallListener { + + private static final Logger LOG = LoggerFactory.getLogger(DynamicToProtoServerCallListener.class); + + private final ServerCall call; + + public DynamicToProtoServerCallListener(final Listener delegate, final ServerCall call) { + super(delegate); + this.call = call; + } + + /** + * dynamicMessage to proto message + */ + @SuppressWarnings("unchecked") + @Override + public void onMessage(final R message) { + Message.Builder builder; + Class t = + DynamicMessageServiceTranslator.getRequestParamsClassMap().get(call.getMethodDescriptor().getFullMethodName()); + try { + builder = (Message.Builder) invokeStaticMethod(t, "newBuilder"); + // dynamicMessage to proto message + String reqData = DynamicJsonMessage.getDataFromDynamicMessage((DynamicMessage) message); + JsonFormat.parser().ignoringUnknownFields().merge(reqData, builder); + if (Objects.isNull(builder)) { + throw new RuntimeException("build json response message is error, newBuilder method is null"); + } + + delegate().onMessage((R) builder.build()); + } catch (Exception e) { + LOG.error("handle json generic request is error", e); + throw Status.INTERNAL.withDescription(e.getMessage()).asRuntimeException(); + } + } + + public static Object invokeStaticMethod(final Class clazz, final String method) { + try { + return MethodUtils.invokeStaticMethod(clazz, method); + } catch (NoSuchMethodException | InvocationTargetException | IllegalAccessException e) { + e.printStackTrace(); + } + return null; + } +} diff --git a/test/plugin/scenarios/grpc-generic-call-scenario/grpc-provider/src/main/java/org/apache/skywalking/apm/testcase/grpc/generic/call/provider/server/dynamic/ProtoToDynamicServerCall.java b/test/plugin/scenarios/grpc-generic-call-scenario/grpc-provider/src/main/java/org/apache/skywalking/apm/testcase/grpc/generic/call/provider/server/dynamic/ProtoToDynamicServerCall.java new file mode 100755 index 000000000..e007fef0d --- /dev/null +++ b/test/plugin/scenarios/grpc-generic-call-scenario/grpc-provider/src/main/java/org/apache/skywalking/apm/testcase/grpc/generic/call/provider/server/dynamic/ProtoToDynamicServerCall.java @@ -0,0 +1,52 @@ +/* + * 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.testcase.grpc.generic.call.provider.server.dynamic; + +import com.google.protobuf.DynamicMessage; +import com.google.protobuf.MessageOrBuilder; +import com.google.protobuf.util.JsonFormat; + +import io.grpc.ForwardingServerCall.SimpleForwardingServerCall; +import io.grpc.ServerCall; +import io.grpc.Status; + +public class ProtoToDynamicServerCall extends SimpleForwardingServerCall { + + protected ProtoToDynamicServerCall(ServerCall delegate) { + super(delegate); + } + + @SuppressWarnings("unchecked") + @Override + public void sendMessage(final P message) { + try { + if (message == null) { + delegate().sendMessage(null); + return; + } + // proto message to dynamicMessage + String jsonFormat = JsonFormat.printer().includingDefaultValueFields().preservingProtoFieldNames() + .print((MessageOrBuilder) message); + DynamicMessage respMessage = DynamicJsonMessage.buildJsonDynamicMessage(jsonFormat); + delegate().sendMessage((P) respMessage); + } catch (Exception e) { + throw Status.INTERNAL.withDescription(e.getMessage()).asRuntimeException(); + } + } +} diff --git a/test/plugin/scenarios/grpc-generic-call-scenario/grpc-provider/src/main/proto/TestService.proto b/test/plugin/scenarios/grpc-generic-call-scenario/grpc-provider/src/main/proto/TestService.proto new file mode 100755 index 000000000..3d3b6f953 --- /dev/null +++ b/test/plugin/scenarios/grpc-generic-call-scenario/grpc-provider/src/main/proto/TestService.proto @@ -0,0 +1,39 @@ +/* + * 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. + * + */ +syntax = "proto3"; + +package org.apache.skywalking.apm.testcase.grpc.proto; + +option java_multiple_files = true; +option java_package = "org.apache.skywalking.apm.testcase.grpc.proto"; + +service TestService { + + rpc unary(RequestData) returns (ResponseData){} + + rpc serverStreaming(RequestData) returns (stream ResponseData){} + +} + +message RequestData { + string data = 1; +} + +message ResponseData { + string data = 1; +} \ No newline at end of file diff --git a/test/plugin/scenarios/grpc-generic-call-scenario/grpc-provider/src/main/resources/application.yml b/test/plugin/scenarios/grpc-generic-call-scenario/grpc-provider/src/main/resources/application.yml new file mode 100644 index 000000000..87862a8f9 --- /dev/null +++ b/test/plugin/scenarios/grpc-generic-call-scenario/grpc-provider/src/main/resources/application.yml @@ -0,0 +1,20 @@ +# 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. + +server: + port: 9999 +grpc: + port: 18080 + diff --git a/test/plugin/scenarios/grpc-generic-call-scenario/pom.xml b/test/plugin/scenarios/grpc-generic-call-scenario/pom.xml new file mode 100644 index 000000000..406f1c397 --- /dev/null +++ b/test/plugin/scenarios/grpc-generic-call-scenario/pom.xml @@ -0,0 +1,93 @@ + + + + + + org.apache.skywalking + grpc-generic-call-scenario + 1.0.0 + pom + + 4.0.0 + + + grpc-consumer + grpc-provider + grpc-dist + + + + UTF-8 + 1.8 + 1.6.0 + 1.6.0 + 2.2.2.RELEASE + 1.18.20 + + 1.2.79 + + + skywalking-grpc-generic-call-scenario + + + + + org.springframework.boot + spring-boot-dependencies + ${spring.boot.version} + pom + import + + + + + io.grpc + grpc-all + ${test.framework.version} + + + + + com.alibaba + fastjson + ${fastjson.version} + + + + + + grpc-generic-call-scenario + + + org.apache.maven.plugins + maven-compiler-plugin + 3.10.1 + + ${compiler.version} + ${compiler.version} + ${project.build.sourceEncoding} + + + + + + + diff --git a/test/plugin/scenarios/grpc-generic-call-scenario/support-version.list b/test/plugin/scenarios/grpc-generic-call-scenario/support-version.list new file mode 100644 index 000000000..2515c3319 --- /dev/null +++ b/test/plugin/scenarios/grpc-generic-call-scenario/support-version.list @@ -0,0 +1,18 @@ +# 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. + +# lists your version here (Contains only the last version number of each minor version.) +1.6.0