Update collectors for latest mqmetric function

Author: ibmmqmet

cmd/mq_aws/exporter.go

@@ -511,7 +511,7 @@ func Collect() error {
 
 	collectStopTime := time.Now()
 	elapsedSecs := int64(collectStopTime.Sub(collectStartTime).Seconds())
-	log.Infof("Collection time = %d secs", elapsedSecs)
+	log.Debugf("Collection time = %d secs", elapsedSecs)
 
 	return err
 

cmd/mq_coll/exporter.go

@@ -395,7 +395,7 @@ func Collect() error {
 
 	collectStopTime := time.Now()
 	elapsedSecs := int64(collectStopTime.Sub(collectStartTime).Seconds())
-	log.Infof("Collection time = %d secs", elapsedSecs)
+	log.Debugf("Collection time = %d secs", elapsedSecs)
 
 	return err
 

cmd/mq_influx/exporter.go

@@ -489,7 +489,7 @@ func Collect(c client.Client) error {
 
 	collectStopTime := time.Now()
 	elapsedSecs := int64(collectStopTime.Sub(collectStartTime).Seconds())
-	log.Infof("Collection time = %d secs", elapsedSecs)
+	log.Debugf("Collection time = %d secs", elapsedSecs)
 
 	return err
 

cmd/mq_json/exporter.go

@@ -484,7 +484,7 @@ func Collect() error {
 
 	collectStopTime := time.Now()
 	elapsedSecs := int64(collectStopTime.Sub(collectStartTime).Seconds())
-	log.Infof("Collection time = %d secs", elapsedSecs)
+	log.Debugf("Collection time = %d secs", elapsedSecs)
 
 	return err
 

cmd/mq_opentsdb/exporter.go

@@ -1,7 +1,7 @@
 package main
 
 /*
-  Copyright (c) IBM Corporation 2016,2020
+  Copyright (c) IBM Corporation 2016,2021
 
   Licensed under the Apache License, Version 2.0 (the "License");
   you may not use this file except in compliance with the License.
@@ -496,7 +496,7 @@ func Collect() error {
 
 	collectStopTime := time.Now()
 	elapsedSecs := int64(collectStopTime.Sub(collectStartTime).Seconds())
-	log.Infof("Collection time = %d secs", elapsedSecs)
+	log.Debugf("Collection time = %d secs", elapsedSecs)
 
 	return err
 }

cmd/mq_prometheus/exporter.go

@@ -27,7 +27,6 @@ and update the various Gauges.
 
 import (
 	"strings"
-	"sync/atomic"
 	"time"
 
 	"github.com/ibm-messaging/mq-golang/v5/ibmmq"
@@ -96,7 +95,7 @@ func (e *exporter) Describe(ch chan<- *prometheus.Desc) {
 
 	// Seeing this twice in succession in the logs is a bit odd. So we'll set things up that
 	// the invocation is not reported during an Unregister operation.
-	if atomic.LoadInt32(&st.collectorSilent) == 0 {
+	if !isCollectorSilent() {
 		log.Infof("IBMMQ Describe started")
 		log.Infof("Platform is %s", platformString)
 	}
@@ -159,7 +158,7 @@ func (e *exporter) Collect(ch chan<- prometheus.Metric) {
 	// If we're not connected, then continue to report a single metric about the qmgr status
 	// This value is created by the mqmetric package even on z/OS which doesn't support
 	// the DIS QMSTATUS command.
-	if atomic.LoadInt32(&st.connectedQMgr) == 0 {
+	if !isConnectedQMgr() {
 		log.Infof("Reporting status as disconnected")
 		if g, ok := qMgrStatusGaugeMap[mqmetric.ATTR_QMGR_STATUS]; ok {
 			// There's no MQQMSTA_STOPPED value defined .All the regular qmgr status
@@ -197,7 +196,7 @@ func (e *exporter) Collect(ch chan<- prometheus.Metric) {
 		pollStatus = false
 		thisPoll := time.Now()
 		elapsed = thisPoll.Sub(lastPoll)
-		if elapsed >= config.cf.PollIntervalDuration || atomic.LoadInt32(&st.firstCollection) == 0 {
+		if elapsed >= config.cf.PollIntervalDuration || !isFirstCollection() {
 			log.Debugf("Polling for object status")
 			lastPoll = thisPoll
 			pollStatus = true
@@ -331,17 +330,16 @@ func (e *exporter) Collect(ch chan<- prometheus.Metric) {
 		// quit out of the collector.
 		switch mqrc {
 		case ibmmq.MQRC_NONE:
-			atomic.StoreInt32(&st.connectedQMgr, 1)
+			setConnectedQMgr(true)
 		case ibmmq.MQRC_UNEXPECTED_ERROR |
 			ibmmq.MQRC_STANDBY_Q_MGR |
 			ibmmq.MQRC_RECONNECT_FAILED:
-			atomic.StoreInt32(&st.collectorEnd, 1)
-			atomic.StoreInt32(&st.connectedQMgr, 0)
+			setCollectorEnd(true)
+			setConnectedQMgr(false)
 		default:
-			atomic.StoreInt32(&st.connectedQMgr, 0)
+			setConnectedQMgr(false)
 		}
 		return
-
 	}
 
 	thisDiscovery := time.Now()
@@ -361,11 +359,11 @@ func (e *exporter) Collect(ch chan<- prometheus.Metric) {
 	// value fields and maps have been updated.
 	//
 	// Now need to set all of the real Gauges with the correct values
-	if atomic.LoadInt32(&st.firstCollection) == 1 {
+	if isFirstCollection() {
 		// Always ignore the first loop through as there might
 		// be accumulated stuff from a while ago, and lead to
 		// a misleading range on graphs.
-		atomic.StoreInt32(&st.firstCollection, 0)
+		setFirstCollection(false)
 	} else {
 
 		for _, cl := range e.metrics.Classes {
@@ -660,7 +658,7 @@ func (e *exporter) Collect(ch chan<- prometheus.Metric) {
 
 	collectStopTime := time.Now()
 	elapsedSecs := int64(collectStopTime.Sub(collectStartTime).Seconds())
-	log.Infof("Collection time = %d secs", elapsedSecs)
+	log.Debugf("Collection time = %d secs", elapsedSecs)
 	if elapsedSecs > defaultScrapeTimeout && !warnedScrapeTimeout {
 		log.Warnf("Collection time has exceeded Prometheus default scrape_timeout value of %d seconds. Ensure you have set a larger value for this job", defaultScrapeTimeout)
 		warnedScrapeTimeout = true

cmd/mq_prometheus/main.go

@@ -26,7 +26,6 @@ import (
 
 	"strings"
 	"sync"
-	"sync/atomic"
 	"time"
 
 	"github.com/ibm-messaging/mq-golang/v5/ibmmq"
@@ -37,17 +36,6 @@ import (
 	log "github.com/sirupsen/logrus"
 )
 
-// These are really booleans but I'm using atomic updates
-// to avoid potential races (though they'd likely be harmless) between
-// the main code and the callbacks
-type status struct {
-	connectedOnce   int32
-	connectedQMgr   int32
-	collectorEnd    int32
-	collectorSilent int32
-	firstCollection int32
-}
-
 var (
 	BuildStamp     string
 	GitCommit      string
@@ -59,7 +47,6 @@ var (
 	collector      prometheus.Collector
 	mutex          sync.RWMutex
 	retryCount     = 0 // Might use this with a maxRetry to force a quit out of collector
-	st             status
 )
 
 func main() {
@@ -76,28 +63,34 @@ func main() {
 	if err != nil {
 		log.Error(err)
 	} else {
-		st.connectedOnce = 0
-		st.connectedQMgr = 0
-		st.collectorEnd = 0
-		st.firstCollection = 0
-		st.collectorSilent = 0
+		setConnectedOnce(false)
+		setConnectedQMgr(false)
+		setCollectorEnd(false)
+		setFirstCollection(false)
+		setCollectorSilent(false)
 
 		// Start the webserver in a separate thread
 		go startServer()
 
 		// This is the main loop that tries to keep the collector connected to a queue manager
 		// even after a failure.
-		for atomic.LoadInt32(&st.collectorEnd) == 0 {
-			log.Debugf("In main loop: qMgrConnected=%d", atomic.LoadInt32(&st.connectedQMgr))
+		for !isCollectorEnd() {
+			log.Debugf("In main loop: qMgrConnected=%v", isConnectedQMgr())
 			err = nil // Start clean on each loop
 
 			// The callback will set this flag to false if there's an error while
 			// processing the messages.
-			if atomic.LoadInt32(&st.connectedQMgr) == 0 {
+			if !isConnectedQMgr() {
 				mutex.Lock()
 				if err == nil {
 					mqmetric.EndConnection()
-					// Connect and open standard queues
+					// Connect and open standard queues. If we're going to manage reconnection from
+					// this collector, then turn off the MQ client automatic option
+					if config.keepRunning {
+						config.cf.CC.SingleConnect = true
+					} else {
+						config.cf.CC.SingleConnect = false
+					}
 					err = mqmetric.InitConnection(config.cf.QMgrName, config.cf.ReplyQ, &config.cf.CC)
 					if err == nil {
 						log.Infoln("Connected to queue manager " + config.cf.QMgrName)
@@ -157,26 +150,26 @@ func main() {
 					allocateAllGauges()
 
 					if collector != nil {
-						atomic.StoreInt32(&st.collectorSilent, 1)
+						setCollectorSilent(true)
 						prometheus.Unregister(collector)
-						atomic.StoreInt32(&st.collectorSilent, 0)
+						setCollectorSilent(false)
 					}
 
 					collector = newExporter()
-					atomic.StoreInt32(&st.firstCollection, 1)
+					setFirstCollection(true)
 					prometheus.MustRegister(collector)
-					atomic.StoreInt32(&st.connectedQMgr, 1)
+					setConnectedQMgr(true)
 
-					if atomic.LoadInt32(&st.connectedOnce) == 0 {
+					if !isConnectedOnce() {
 						startChannel <- true
-						atomic.StoreInt32(&st.connectedOnce, 1)
+						setConnectedOnce(true)
 					}
 				} else {
-					if atomic.LoadInt32(&st.connectedOnce) == 0 || !config.keepRunning {
+					if !isConnectedOnce() || !config.keepRunning {
 						// If we've never successfully connected, then exit instead
 						// of retrying as it probably means a config error
 						log.Errorf("Connection to %s has failed. %v", config.cf.QMgrName, err)
-						atomic.StoreInt32(&st.collectorEnd, 1)
+						setCollectorEnd(true)
 					} else {
 						log.Debug("Sleeping a bit after a failure")
 						retryCount++
@@ -197,7 +190,6 @@ func main() {
 	} else {
 		os.Exit(0)
 	}
-
 }
 
 func startServer() {
@@ -268,7 +260,7 @@ func stopServer() {
 	if err != nil {
 		log.Errorf("Failed to shutdown metrics server: %v", err)
 	}
-	atomic.StoreInt32(&st.collectorEnd, 1)
+	setCollectorEnd(true)
 }
 
 /*

cmd/mq_prometheus/status.go

@@ -0,0 +1,124 @@
+package main
+
+/*
+  Copyright (c) IBM Corporation 2021
+
+  Licensed 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
+
+   Contributors:
+     Mark Taylor - Initial Contribution
+*/
+
+/*
+This file manages the state between different threads of the collector. It exposes
+the state via getter/setter functions which use atomic read/write operations. We also
+map between integer values internally and boolean state which is all we really care about.
+*/
+
+import (
+	"sync/atomic"
+)
+
+// These are really booleans but I'm using atomic updates
+// to avoid potential races (though they'd likely be harmless) between
+// the main code and the callbacks
+type status struct {
+	connectedOnce   int32
+	connectedQMgr   int32
+	collectorEnd    int32
+	collectorSilent int32
+	firstCollection int32
+}
+
+var (
+	st status
+)
+
+func isConnectedQMgr() bool {
+	b := atomic.LoadInt32(&st.connectedQMgr)
+	if b == 0 {
+		return false
+	}
+	return true
+}
+
+func setConnectedQMgr(b bool) {
+	if b {
+		atomic.StoreInt32(&st.connectedQMgr, 1)
+	} else {
+		atomic.StoreInt32(&st.connectedQMgr, 0)
+	}
+}
+
+func isCollectorEnd() bool {
+	b := atomic.LoadInt32(&st.collectorEnd)
+	if b == 0 {
+		return false
+	}
+	return true
+}
+
+func setCollectorEnd(b bool) {
+	if b {
+		atomic.StoreInt32(&st.collectorEnd, 1)
+	} else {
+		atomic.StoreInt32(&st.collectorEnd, 0)
+	}
+}
+
+func isFirstCollection() bool {
+	b := atomic.LoadInt32(&st.firstCollection)
+	if b == 0 {
+		return false
+	}
+	return true
+}
+
+func setFirstCollection(b bool) {
+	if b {
+		atomic.StoreInt32(&st.firstCollection, 1)
+	} else {
+		atomic.StoreInt32(&st.firstCollection, 0)
+	}
+}
+
+func isCollectorSilent() bool {
+	b := atomic.LoadInt32(&st.collectorSilent)
+	if b == 0 {
+		return false
+	}
+	return true
+}
+
+func setCollectorSilent(b bool) {
+	if b {
+		atomic.StoreInt32(&st.collectorSilent, 1)
+	} else {
+		atomic.StoreInt32(&st.collectorSilent, 0)
+	}
+}
+
+func setConnectedOnce(b bool) {
+	if b {
+		atomic.StoreInt32(&st.connectedOnce, 1)
+	} else {
+		atomic.StoreInt32(&st.connectedOnce, 0)
+	}
+}
+
+func isConnectedOnce() bool {
+	b := atomic.LoadInt32(&st.connectedOnce)
+	if b == 0 {
+		return false
+	}
+	return true
+}