Merge pull request #116 from Cinimex-Informatica/feature/issue20_add_timeouts

Get message timeout added to config file

Author: willardgibbs

README.md

@@ -91,6 +91,9 @@ qmgrConnectionParams:
   password: mqm
 # Use MQCSP for connection?
   mqscp: false
+# How long to wait until metrics are published by queue manager (milliseconds).
+# Value must be at least 10000 (periodicity with which metrics are published by MQ).
+  connTimeout: 12000
 
 # Prometheus connection information -------------------------------
 prometheusEndpointParams:

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

@@ -25,6 +25,7 @@ public class Config {
     private String user;
     private String password;
     private boolean mqscp;
+    private int connTimeout;
     private int endpPort;
     private String endpURL;
     private ArrayList<String> queues;
@@ -55,6 +56,7 @@ public Config(String path) {
         this.user = (String) qmgrConnectionParams.get("user");
         this.password = (String) qmgrConnectionParams.get("password");
         this.mqscp = (boolean) qmgrConnectionParams.get("mqscp");
+        this.connTimeout = (Integer) qmgrConnectionParams.get("connTimeout");
         queues = (ArrayList<String>) config.get("queues");
         listeners = (ArrayList<String>) config.get("listeners");
         channels = (ArrayList<String>) config.get("channels");
@@ -114,6 +116,10 @@ public String getPassword() {
         return password;
     }
 
+    public int getConnTimeout() {
+        return connTimeout;
+    }
+
     public int getEndpPort() {
         return endpPort;
     }

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

@@ -64,7 +64,8 @@ public static void main(String[] args) {
 
         MetricsManager.initMetrics(elements, monitoringTypes);
         MQSubscriberManager manager = new MQSubscriberManager(config.getQmgrHost(), config.getQmgrPort(), config.getQmgrChannel(), config.getQmgrName(), config.getUser(), config.getPassword(), config.useMqscp());
-        manager.runSubscribers(elements, objects, config.sendPCFCommands(), config.usePCFWildcards(), config.getScrapeInterval());
+        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);
         } catch (IOException e) {

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

@@ -44,11 +44,12 @@ public MQSubscriberManager(String host, int port, String channel, String qmName,
      * @param sendPCFCommands - this flag indicates, if we should send additional PCF commands (To get queues max depth, channels and listeners statuses).
      * @param usePCFWildcards - this flag indicates, if we should use wildcards (uses only 1 connection per MQObject type, but longer response processing).
      * @param interval        - interval in seconds, at which additional PCF commands are sent.
+     * @param timeout         - timeout for MQGET operation (milliseconds).
      */
-    public void runSubscribers(List<PCFElement> elements, List<MQObject> objects, boolean sendPCFCommands, boolean usePCFWildcards, int interval) {
+    public void runSubscribers(List<PCFElement> elements, List<MQObject> objects, boolean sendPCFCommands, boolean usePCFWildcards, int interval, int timeout) {
         logger.info("Launching subscribers...");
         subscribers = new ArrayList<>();
-        addTopicSubscribers(elements, objects);
+        addTopicSubscribers(elements, objects, timeout);
         if (sendPCFCommands) {
             if (usePCFWildcards) {
                 EnumMap<MQObject.MQType, ArrayList<MQObject>> groups = groupMQObjects(objects);
@@ -83,15 +84,16 @@ public void stopSubscribers() {
      *
      * @param elements - list with PCFElements, received from MQ.
      * @param objects  - list with monitored MQ objects.
+     * @param timeout  - timeout for MQGET operation (milliseconds).
      */
-    private void addTopicSubscribers(List<PCFElement> elements, List<MQObject> objects) {
+    private void addTopicSubscribers(List<PCFElement> elements, List<MQObject> objects, int timeout) {
         for (PCFElement element : elements) {
             if (element.requiresMQObject()) {
                 for (MQObject object : objects) {
-                    addTopicSubscriber(object, element);
+                    addTopicSubscriber(object, element, timeout);
                 }
             } else {
-                addTopicSubscriber(element);
+                addTopicSubscriber(element, timeout);
             }
         }
     }
@@ -101,13 +103,14 @@ private void addTopicSubscribers(List<PCFElement> elements, List<MQObject> objec
      *
      * @param object  - monitored MQ object.
      * @param element - PCFElement, received from MQ.
+     * @param timeout - timeout for MQGET operation (milliseconds).
      */
-    private void addTopicSubscriber(MQObject object, PCFElement element) {
+    private void addTopicSubscriber(MQObject object, PCFElement element, int timeout) {
         if (object.getType().equals(MQObject.MQType.QUEUE)) {
             PCFElement objElement = new PCFElement(element.getTopicString(), element.getRows());
             objElement.formatTopicString(object.getName());
             try {
-                subscribers.add(new MQTopicSubscriber(objElement, queueManagerName, connectionProperties, queueManagerName, object.getName()));
+                subscribers.add(new MQTopicSubscriber(objElement, queueManagerName, connectionProperties, timeout, queueManagerName, object.getName()));
             } catch (MQException e) {
                 logger.error("Error during creating topic subscriber: ", e);
             }
@@ -118,10 +121,11 @@ private void addTopicSubscriber(MQObject object, PCFElement element) {
      * Adds topic subscriber
      *
      * @param element - PCFElement, received from MQ.
+     * @param timeout - timeout for MQGET operation (milliseconds).
      */
-    private void addTopicSubscriber(PCFElement element) {
+    private void addTopicSubscriber(PCFElement element, int timeout) {
         try {
-            subscribers.add(new MQTopicSubscriber(element, queueManagerName, connectionProperties, queueManagerName));
+            subscribers.add(new MQTopicSubscriber(element, queueManagerName, connectionProperties, timeout, queueManagerName));
         } catch (MQException e) {
             logger.error("Error during creating topic subscriber: ", e);
         }

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

@@ -25,20 +25,23 @@ public class MQTopicSubscriber extends MQSubscriber {
     private PCFElement element;
     private MQConnection connection;
     private String[] labels;
+    private int timeout;
     private boolean isRunning;
 
     /**
      * Subscriber constructor
      *
-     * @param element - PCF message data, which is required for parsing statistics.
-     * @param labels  - labels array, which should be used for metrics.
+     * @param element              - PCF message data, which is required for parsing statistics.
+     * @param queueManagerName     - queue manager name.
+     * @param connectionProperties - structure, filled with connection properties.
+     * @param timeout              - timeout for MQGET operation (milliseconds).
+     * @param labels               - labels array, which should be used for metrics.
      */
-    public MQTopicSubscriber(PCFElement element, String queueManagerName, Hashtable<String, Object> connectionProperties, String... labels) throws MQException {
+    public MQTopicSubscriber(PCFElement element, String queueManagerName, Hashtable<String, Object> connectionProperties, int timeout, String... labels) throws MQException {
         this.element = element;
-        if (connection == null) {
-            connection = new MQConnection();
-            connection.establish(queueManagerName, connectionProperties);
-        }
+        this.timeout = timeout;
+        this.connection = new MQConnection();
+        this.connection.establish(queueManagerName, connectionProperties);
         this.labels = labels;
     }
 
@@ -86,7 +89,7 @@ public void run() {
             topic = connection.createTopic(element.getTopicString());
             MQGetMessageOptions gmo = new MQGetMessageOptions();
             gmo.options = MQConstants.MQGMO_WAIT | MQConstants.MQGMO_COMPLETE_MSG;
-            gmo.waitInterval = 12000;
+            gmo.waitInterval = timeout;
             isRunning = true;
             while (isRunning) {
                 scrapeMetrics(gmo);

src/main/resources/exporter_config.yaml

@@ -16,6 +16,9 @@ qmgrConnectionParams:
   password: mqm
 # Use MQCSP for connection?
   mqscp: false
+# How long to wait until metrics are published by queue manager (milliseconds).
+# Value must be at least 10000 (periodicity with which metrics are published by MQ).
+  connTimeout: 12000
 
 # Prometheus connection information -------------------------------
 prometheusEndpointParams: