Merge pull request #130 from Cinimex-Informatica/feature/issue17_shutdown_handling

ShutdownHook implemented

Author: echerniak

src/main/java/ru/cinimex/exporter/ExporterLauncher.java

@@ -31,6 +31,8 @@ public class ExporterLauncher {
     private static final Logger logger = LogManager.getLogger(ExporterLauncher.class);
     private static final String TOPIC_STRING = "$SYS/MQ/INFO/QMGR/%s/Monitor/METADATA/CLASSES";
     private static final int GMO = MQConstants.MQGMO_WAIT | MQConstants.MQGMO_COMPLETE_MSG | MQConstants.MQGMO_SYNCPOINT;
+    private static  MQSubscriberManager manager;
+    private static HTTPServer server;
 
     public static void main(String[] args) {
         if (args.length == 0) {
@@ -39,6 +41,7 @@ public static void main(String[] args) {
         }
         Config config = new Config(args[0]);
 
+        createShutdownHook();
         ArrayList<PCFElement> elements = getAllPublishedMetrics(config);
         ArrayList<MQObject.MQType> monitoringTypes = new ArrayList<>();
         ArrayList<MQObject> objects = new ArrayList<>();
@@ -63,11 +66,11 @@ public static void main(String[] args) {
         }
 
         MetricsManager.initMetrics(elements, monitoringTypes);
-        MQSubscriberManager manager = new MQSubscriberManager(config);
+        manager = new MQSubscriberManager(config);
         manager.runSubscribers(elements, objects, config.sendPCFCommands(), config.usePCFWildcards(),
                 config.getScrapeInterval(), config.getConnTimeout());
         try {
-            new HTTPServer(new InetSocketAddress("0.0.0.0", config.getEndpPort()), config.getEndpURL(), Registry.getRegistry(), false);
+            server = new HTTPServer(new InetSocketAddress("0.0.0.0", config.getEndpPort()), config.getEndpURL(), Registry.getRegistry(), false);
         } catch (IOException e) {
             logger.error("Error occurred during expanding endpoint for Prometheus: ", e);
         }
@@ -123,6 +126,27 @@ private static ArrayList<PCFElement> getAllPublishedMetrics(Config config) {
         return elements;
     }
 
+    private static void createShutdownHook() {
+        Runtime.getRuntime().addShutdownHook(new Thread(() -> {
+            logger.info("Exporter finishes all activities...");
+            if (manager != null) {
+                try {
+                    logger.debug("Stopping subscribers... (it may take some time, please, be patient)");
+                    manager.stopSubscribers();
+                } catch (InterruptedException e) {
+                    logger.error("Error occurred during stopping subscribers: ", e);
+                }
+            }
+
+            if (server != null) {
+                logger.debug("Stopping HTTP server...");
+                server.stop();
+            }
+            logger.info("Goodbye!");
+            LogManager.shutdown();
+        }));
+    }
+
     private static MQMessage getEmptyMessage() {
         MQMessage message = new MQMessage();
         message.messageId = null;

src/main/java/ru/cinimex/exporter/mq/MQPCFSubscriber.java

@@ -146,8 +146,12 @@ private void updateWithDirectPCFCommand(List<String> objectNames) {
     public void stopProcessing() {
         isRunning = false;
         try {
-            agent.disconnect();
-            connection.close();
+            if (agent != null) {
+                agent.disconnect();
+            }
+            if (connection != null) {
+                connection.close();
+            }
         } catch (MQException e) {
             logger.error("Error occurred during stopping PCF subscriber: ", e);
         }

src/main/java/ru/cinimex/exporter/mq/MQSubscriberManager.java

@@ -20,6 +20,7 @@ public class MQSubscriberManager {
     private String queueManagerName;
     private ArrayList<MQSubscriber> subscribers;
     private ScheduledExecutorService executor;
+    private int timeout;
 
     /**
      * Constructor sets params for connecting to target queue manager.
@@ -44,6 +45,7 @@ public MQSubscriberManager(Config config) {
     public void runSubscribers(List<PCFElement> elements, List<MQObject> objects, boolean sendPCFCommands, boolean usePCFWildcards, int interval, int timeout) {
         logger.info("Launching subscribers...");
         subscribers = new ArrayList<>();
+        this.timeout = timeout;
         addTopicSubscribers(elements, objects, timeout);
         if (sendPCFCommands) {
             if (usePCFWildcards) {
@@ -65,13 +67,20 @@ public void runSubscribers(List<PCFElement> elements, List<MQObject> objects, bo
 
     }
 
-    public void stopSubscribers() {
+    /**
+     * Stops all running subscribers in managed mode. All connections will be closed, all threads will be finished.
+     * @throws InterruptedException
+     */
+    public void stopSubscribers() throws InterruptedException {
         if (executor != null) {
             executor.shutdown();
         }
         for (MQSubscriber subscriber : subscribers) {
             subscriber.stopProcessing();
         }
+        for (MQSubscriber subscriber : subscribers) {
+            subscriber.join(timeout);
+        }
     }
 
     /**

src/main/java/ru/cinimex/exporter/mq/MQTopicSubscriber.java

@@ -98,7 +98,6 @@ public void run() {
         } catch (MQException e) {
             logger.error("Error occurred during establishing connection with topic {}", element.getTopicString(), e);
         } finally {
-            System.out.println("Finishing topic work!");
             try {
                 if (topic != null && topic.isOpen()) {
                     topic.close();

src/main/resources/log4j2.properties

@@ -1,5 +1,6 @@
 status = warn
 name= exporter_log_configuration
+shutdownHook = disable
 
 # Give directory path where log files should get stored
 property.basePath = ./log/