Add plugin to support NATS client

.github/workflows/plugins-test.9999.yaml

@@ -0,0 +1,77 @@
+# 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.
+
+name: Test
+
+on:
+  pull_request:
+    paths:
+      - '.github/workflows/plugins-*.yaml'
+      - 'apm-application-toolkit/**'
+      - 'apm-commons/**'
+      - 'apm-protocol/**'
+      - 'apm-sniffer/**'
+      - 'test/plugin/**'
+      - '**/pom.xml'
+      - '!**.md'
+  push:
+    branches:
+      - test/ci/*
+
+concurrency:
+  group: plugins-3-${{ github.event.pull_request.number || github.ref }}
+  cancel-in-progress: true
+
+jobs:
+  build:
+    name: Build
+    runs-on: ubuntu-latest
+    timeout-minutes: 30
+    steps:
+      - uses: actions/checkout@v2
+        with:
+          submodules: true
+      - name: Build
+        uses: ./.github/actions/build
+
+  test:
+    needs: [ build ]
+    name: ${{ matrix.case }}
+    runs-on: ubuntu-latest
+    timeout-minutes: 90
+    strategy:
+      matrix:
+        case:
+          - nats-client-scenario
+    steps:
+      - uses: actions/checkout@v2
+        with:
+          submodules: true
+      - uses: actions/setup-java@v2
+        with:
+          distribution: adopt
+          java-version: 8
+      - name: Install Oracle Libs
+        if: matrix.case == 'oracle-scenario'
+        run: |
+          mkdir -p skywalking-agent/plugins
+          curl -O https://skyapm.github.io/ci-assist/jars/ojdbc14-10.2.0.4.0.jar
+          curl -L -o ./skywalking-agent/plugins/apm-oracle-10.x-plugin-2.0.0.jar https://github.com/SkyAPM/java-plugin-extensions/releases/download/2.0.0/apm-oracle-10.x-plugin-2.0.0.jar
+          ./mvnw -q --batch-mode install:install-file -Dfile=ojdbc14-10.2.0.4.0.jar -DgroupId=com.oracle -DartifactId=ojdbc14 -Dversion=10.2.0.4.0 -Dpackaging=jar
+      - name: Run Plugin Test
+        uses: ./.github/actions/run
+        with:
+          test_case: ${{ matrix.case }}

CHANGES.md

@@ -18,6 +18,7 @@ Release Notes.
 * Force the injected high-priority classes in order to avoid NoClassDefFoundError.
 * Plugin to support xxl-job 2.3.x.
 * Add plugin to support Micronaut(HTTP Client/Server) 3.2.x-3.6.x
+* Add plugin to support NATS Java client
 
 #### Documentation
 

apm-protocol/apm-network/src/main/java/org/apache/skywalking/apm/network/trace/component/ComponentsDefine.java

@@ -227,4 +227,6 @@ public class ComponentsDefine {
 
     public static final OfficialComponent MICRONAUT = new OfficialComponent(131, "Micronaut");
 
+    public static final OfficialComponent NATS = new OfficialComponent(132, "Nats");
+
 }

apm-sniffer/apm-sdk-plugin/nats-client-plugin/pom.xml

@@ -0,0 +1,47 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+  ~ 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.
+  ~
+  -->
+
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
+    <parent>
+        <artifactId>apm-sdk-plugin</artifactId>
+        <groupId>org.apache.skywalking</groupId>
+        <version>8.12.0-SNAPSHOT</version>
+    </parent>
+
+    <modelVersion>4.0.0</modelVersion>
+
+    <artifactId>nats-client-plugin</artifactId>
+    <packaging>jar</packaging>
+
+    <properties>
+        <maven.compiler.source>8</maven.compiler.source>
+        <maven.compiler.target>8</maven.compiler.target>
+    </properties>
+
+    <dependencies>
+        <dependency>
+            <groupId>io.nats</groupId>
+            <artifactId>jnats</artifactId>
+            <version>2.15.6</version>
+            <scope>provided</scope>
+        </dependency>
+    </dependencies>
+
+</project>

apm-sniffer/apm-sdk-plugin/nats-client-plugin/src/main/java/org/apache/skywalking/apm/plugin/nats/client/CreateDispatcherInterceptor.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.nats.client;
+
+import io.nats.client.Connection;
+import io.nats.client.MessageHandler;
+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;
+import java.lang.reflect.Method;
+import static org.apache.skywalking.apm.plugin.nats.client.NatsCommons.buildTraceMsgHandler;
+
+public class CreateDispatcherInterceptor implements InstanceMethodsAroundInterceptor {
+
+    @Override
+    public void beforeMethod(EnhancedInstance objInst, Method method, Object[] allArguments, Class<?>[] argumentsTypes, MethodInterceptResult result) throws Throwable {
+        allArguments[0] = buildTraceMsgHandler((MessageHandler) allArguments[0], (Connection) objInst);
+    }
+
+    @Override
+    public Object afterMethod(EnhancedInstance objInst, Method method, Object[] allArguments, Class<?>[] argumentsTypes, Object ret) throws Throwable {
+        return ret;
+    }
+
+    @Override
+    public void handleMethodException(EnhancedInstance objInst, Method method, Object[] allArguments, Class<?>[] argumentsTypes, Throwable t) {
+    }
+
+}

apm-sniffer/apm-sdk-plugin/nats-client-plugin/src/main/java/org/apache/skywalking/apm/plugin/nats/client/CreateSubInterceptor.java

@@ -0,0 +1,47 @@
+/*
+ *   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.nats.client;
+
+import io.nats.client.Connection;
+import io.nats.client.MessageHandler;
+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;
+
+import java.lang.reflect.Method;
+
+import static org.apache.skywalking.apm.plugin.nats.client.NatsCommons.buildTraceMsgHandler;
+
+public class CreateSubInterceptor implements InstanceMethodsAroundInterceptor {
+    @Override
+    public void beforeMethod(EnhancedInstance objInst, Method method, Object[] allArguments, Class<?>[] argumentsTypes, MethodInterceptResult result) throws Throwable {
+        allArguments[3] = buildTraceMsgHandler((MessageHandler) allArguments[3], (Connection) objInst.getSkyWalkingDynamicField());
+    }
+
+    @Override
+    public Object afterMethod(EnhancedInstance objInst, Method method, Object[] allArguments, Class<?>[]
+            argumentsTypes, Object ret) throws Throwable {
+        return ret;
+    }
+
+    @Override
+    public void handleMethodException(EnhancedInstance objInst, Method method, Object[] allArguments, Class<?>[]
+            argumentsTypes, Throwable t) {
+
+    }
+}

apm-sniffer/apm-sdk-plugin/nats-client-plugin/src/main/java/org/apache/skywalking/apm/plugin/nats/client/DeliverReplyInterceptor.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 org.apache.skywalking.apm.plugin.nats.client;
+
+import io.nats.client.Connection;
+import io.nats.client.Message;
+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.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;
+
+import java.lang.reflect.Method;
+
+import static org.apache.skywalking.apm.plugin.nats.client.NatsCommons.createIncomingSpan;
+
+public class DeliverReplyInterceptor implements InstanceMethodsAroundInterceptor {
+
+    @Override
+    public void beforeMethod(EnhancedInstance objInst, Method method, Object[] allArguments, Class<?>[] argumentsTypes, MethodInterceptResult result) throws Throwable {
+        createIncomingSpan((Message) allArguments[0], (Connection) objInst);
+    }
+
+    @Override
+    public Object afterMethod(EnhancedInstance objInst, Method method, Object[] allArguments, Class<?>[] argumentsTypes, Object ret) throws Throwable {
+        if (ContextManager.isActive()) {
+            AbstractSpan span = ContextManager.activeSpan();
+            ContextManager.stopSpan(span);
+        }
+        return ret;
+    }
+
+    @Override
+    public void handleMethodException(EnhancedInstance objInst, Method method, Object[] allArguments, Class<?>[] argumentsTypes, Throwable t) {
+        if (ContextManager.isActive()) {
+            AbstractSpan span = ContextManager.activeSpan();
+            span.log(t).errorOccurred();
+            ContextManager.stopSpan(span);
+        }
+    }
+}

apm-sniffer/apm-sdk-plugin/nats-client-plugin/src/main/java/org/apache/skywalking/apm/plugin/nats/client/NatsClientPluginConfig.java

@@ -0,0 +1,38 @@
+/*
+ * 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.nats.client;
+
+import org.apache.skywalking.apm.agent.core.boot.PluginConfig;
+
+public class NatsClientPluginConfig {
+
+    public static class Plugin {
+        @PluginConfig(root = NatsClientPluginConfig.class)
+        public static class NatsClient {
+            /**
+             * Nats publish message asynchronously , it put message to local queue ,
+             * then write message to network by call flush method in another thread .
+             * This config term indicate whether collect complete trace .
+             * If set to true ,the plugin will trace enqueue , flush . Otherwise, only enqueue
+             * Notice , If set true . will generate a lot of Span (one span for a message). These spans are not released util call flush
+             */
+            public static boolean ENABLE_FULL_TRACE = false;
+        }
+    }
+}