Merge pull request #297 from wu-sheng/feature/jvm-shutdown-hook

Add jvm shutdown hook

apm-commons/apm-datacarrier/src/main/java/org/skywalking/apm/commons/datacarrier/consumer/ConsumerPool.java

@@ -20,6 +20,7 @@ public class ConsumerPool<T> {
         this(channels, num);
         for (int i = 0; i < num; i++) {
             consumerThreads[i] = new ConsumerThread("DataCarrier.Consumser." + i + ".Thread", getNewConsumerInstance(consumerClass));
+            consumerThreads[i].setDaemon(true);
         }
     }
 

apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/boot/BootService.java

@@ -13,4 +13,6 @@ public interface BootService {
     void boot() throws Throwable;
 
     void afterBoot() throws Throwable;
+
+    void shutdown() throws Throwable;
 }

apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/boot/ServiceManager.java

@@ -27,6 +27,16 @@ public enum ServiceManager {
         afterBoot();
     }
 
+    public void shutdown() {
+        for (BootService service : bootedServices.values()) {
+            try {
+                service.shutdown();
+            } catch (Throwable e) {
+                logger.error(e, "ServiceManager try to shutdown [{}] fail.", service.getClass().getName());
+            }
+        }
+    }
+
     private Map<Class, BootService> loadAllServices() {
         HashMap<Class, BootService> bootedServices = new HashMap<Class, BootService>();
         Iterator<BootService> serviceIterator = load().iterator();

apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/context/ContextManager.java

@@ -162,6 +162,10 @@ public class ContextManager implements TracingContextListener, BootService, Igno
 
     }
 
+    @Override public void shutdown() throws Throwable {
+
+    }
+
     @Override
     public void afterFinished(TraceSegment traceSegment) {
         CONTEXT.remove();

apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/jvm/JVMService.java

@@ -62,6 +62,12 @@ public class JVMService implements BootService, Runnable {
 
     }
 
+    @Override
+    public void shutdown() throws Throwable {
+        collectMetricFuture.cancel(true);
+        sendMetricFuture.cancel(true);
+    }
+
     @Override
     public void run() {
         if (RemoteDownstreamConfig.Agent.APPLICATION_ID != DictionaryUtil.nullValue()

apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/remote/AppAndServiceRegisterClient.java

@@ -77,6 +77,11 @@ public class AppAndServiceRegisterClient implements BootService, GRPCChannelList
         TracingContext.ListenerManager.add(this);
     }
 
+    @Override
+    public void shutdown() throws Throwable {
+        applicationRegisterFuture.cancel(true);
+    }
+
     @Override
     public void run() {
         if (CONNECTED.equals(status)) {

apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/remote/CollectorDiscoveryService.java

@@ -1,6 +1,7 @@
 package org.skywalking.apm.agent.core.remote;
 
 import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledFuture;
 import java.util.concurrent.TimeUnit;
 import org.skywalking.apm.agent.core.boot.BootService;
 import org.skywalking.apm.agent.core.conf.Config;
@@ -11,6 +12,8 @@ import org.skywalking.apm.agent.core.conf.Config;
  * @author wusheng
  */
 public class CollectorDiscoveryService implements BootService {
+    private ScheduledFuture<?> future;
+
     @Override
     public void beforeBoot() throws Throwable {
 
@@ -18,7 +21,7 @@ public class CollectorDiscoveryService implements BootService {
 
     @Override
     public void boot() throws Throwable {
-        Executors.newSingleThreadScheduledExecutor()
+        future = Executors.newSingleThreadScheduledExecutor()
             .scheduleAtFixedRate(new DiscoveryRestServiceClient(), 0,
                 Config.Collector.DISCOVERY_CHECK_INTERVAL, TimeUnit.SECONDS);
     }
@@ -27,4 +30,9 @@ public class CollectorDiscoveryService implements BootService {
     public void afterBoot() throws Throwable {
 
     }
+
+    @Override
+    public void shutdown() throws Throwable {
+        future.cancel(true);
+    }
 }

apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/remote/GRPCChannelManager.java

@@ -49,6 +49,12 @@ public class GRPCChannelManager implements BootService, Runnable {
 
     }
 
+    @Override
+    public void shutdown() throws Throwable {
+        connectCheckFuture.cancel(true);
+        managedChannel.shutdownNow();
+    }
+
     @Override
     public void run() {
         if (reconnect) {

apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/remote/TraceSegmentServiceClient.java

@@ -49,6 +49,11 @@ public class TraceSegmentServiceClient implements BootService, IConsumer<TraceSe
         TracingContext.ListenerManager.add(this);
     }
 
+    @Override
+    public void shutdown() throws Throwable {
+        carrier.shutdownConsumers();
+    }
+
     @Override
     public void init() {
 

apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/sampling/SamplingService.java

@@ -61,6 +61,11 @@ public class SamplingService implements BootService {
 
     }
 
+    @Override
+    public void shutdown() throws Throwable {
+        scheduledFuture.cancel(true);
+    }
+
     /**
      * @return true, if sampling mechanism is on, and get the sampling factor successfully.
      */

apm-sniffer/apm-agent/src/main/java/org/skywalking/apm/agent/SkyWalkingAgent.java

@@ -44,6 +44,12 @@ public class SkyWalkingAgent {
 
         ServiceManager.INSTANCE.boot();
 
+        Runtime.getRuntime().addShutdownHook(new Thread(new Runnable() {
+            @Override public void run() {
+                ServiceManager.INSTANCE.shutdown();
+            }
+        }, "skywalking service shutdown thread"));
+
         new AgentBuilder.Default().type(pluginFinder.buildMatch()).transform(new AgentBuilder.Transformer() {
             @Override
             public DynamicType.Builder<?> transform(DynamicType.Builder<?> builder, TypeDescription typeDescription,