Add RESET QSTATS for z/OS monitoring

Author: ibmmqmet

CHANGELOG.md

@@ -1,6 +1,12 @@
 # Changelog
 Newest updates are at the top of this file.
-## July 11 2018 - v4.0.8
+
+## July 22 2019 - v4.0.9
+* mqmetric - Support RESET QSTATS on z/OS queue manager
+* mqmetric - Add a Logger class to enable debug output
+* mqmetric - Improve some error reports
+
+## July 11 2019 - v4.0.8
 * Update for MQ 9.1.3 - No new API function introduced
 * mqmetric - Fix leak in subscriptions after rediscovery
 * mqmetric - Add USAGE as a queue label for selection by xmitq

mqmetric/discover.go

@@ -81,10 +81,10 @@ type AllMetrics struct {
 }
 
 type QInfo struct {
-	MaxDepth        int64
-	Usage           int64
-	exists          bool // Used during rediscovery
-	firstCollection bool // To indicate discard needed of first stat
+	AttrMaxDepth    int64 // The queue attribute value. Not the max depth reported by RESET QSTATS
+	AttrUsage       int64 // Normal or XMITQ
+	exists          bool  // Used during rediscovery
+	firstCollection bool  // To indicate discard needed of first stat
 }
 
 // QMgrMapKey can never be a real object name and is therefore useful in
@@ -559,7 +559,7 @@ func discoverQueues(monitoredQueuePatterns string) error {
 			if qInfoElem, ok = qInfoMap[qName]; !ok {
 				qInfoElem = new(QInfo)
 			}
-			qInfoElem.MaxDepth = defaultMaxQDepth
+			qInfoElem.AttrMaxDepth = defaultMaxQDepth
 			qInfoElem.exists = true
 			qInfoMap[qName] = qInfoElem
 		}
@@ -696,7 +696,8 @@ func inquireObjects(objectPatternsList string, objectType int32) ([]string, erro
 				truncation = false
 				cfh, offset := ibmmq.ReadPCFHeader(buf)
 				if cfh.CompCode != ibmmq.MQCC_OK {
-					return objectList, fmt.Errorf("PCF command failed with CC %s [%d] RC %s [%d]",
+					return objectList, fmt.Errorf("PCF command %s [%d] failed with CC %s [%d] RC %s [%d]",
+						ibmmq.MQItoString("CMD", int(cfh.Command)), cfh.Command,
 						ibmmq.MQItoString("CC", int(cfh.CompCode)), cfh.CompCode,
 						ibmmq.MQItoString("RC", int(cfh.Reason)), cfh.Reason)
 				} else {

mqmetric/log.go

@@ -0,0 +1,48 @@
+package mqmetric
+
+/*
+  Copyright (c) IBM Corporation 2016, 2019
+
+  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 language governing permissions and
+  limitations under the License.
+
+   Contributors:
+     Mark Taylor - Initial Contribution
+*/
+
+type Logger struct {
+	Debug func(string, ...interface{})
+	Info  func(string, ...interface{})
+	Error func(string, ...interface{})
+}
+
+var logger *Logger = nil
+
+func SetLogger(l *Logger) {
+	logger = l
+}
+
+func logDebug(format string, v ...interface{}) {
+	if logger != nil && logger.Debug != nil {
+		logger.Debug(format, v...)
+	}
+}
+func logInfo(format string, v ...interface{}) {
+	if logger != nil && logger.Info != nil {
+		logger.Info(format, v...)
+	}
+}
+func logError(format string, v ...interface{}) {
+	if logger != nil && logger.Info != nil {
+		logger.Error(format, v...)
+	}
+}

mqmetric/mqif.go

@@ -26,7 +26,6 @@ don't need to repeat common setups eg of MQMD or MQSD structures.
 
 import (
 	"fmt"
-
 	"github.com/ibm-messaging/mq-golang/ibmmq"
 )
 
@@ -49,6 +48,7 @@ var (
 
 	usePublications = true
 	useStatus       = false
+	useResetQStats  = false
 )
 
 type ConnectionConfig struct {
@@ -59,6 +59,7 @@ type ConnectionConfig struct {
 
 	UsePublications bool
 	UseStatus       bool
+	UseResetQStats  bool
 }
 
 /*
@@ -90,6 +91,7 @@ func InitConnection(qMgrName string, replyQ string, cc *ConnectionConfig) error
 		gocno.SecurityParms = gocsp
 	}
 
+	logDebug("Connecting to queue manager %s", qMgrName)
 	qMgr, err = ibmmq.Connx(qMgrName, gocno)
 	if err == nil {
 		qmgrConnected = true
@@ -117,6 +119,7 @@ func InitConnection(qMgrName string, replyQ string, cc *ConnectionConfig) error
 		if err == nil {
 			selectors := []int32{ibmmq.MQCA_Q_MGR_NAME,
 				ibmmq.MQIA_COMMAND_LEVEL,
+				ibmmq.MQIA_PERFORMANCE_EVENT,
 				ibmmq.MQIA_PLATFORM}
 
 			v, err = qMgrObject.InqMap(selectors)
@@ -126,6 +129,12 @@ func InitConnection(qMgrName string, replyQ string, cc *ConnectionConfig) error
 				commandLevel = v[ibmmq.MQIA_COMMAND_LEVEL].(int32)
 				if platform == ibmmq.MQPL_ZOS {
 					usePublications = false
+					useResetQStats = cc.UseResetQStats
+					evEnabled := v[ibmmq.MQIA_PERFORMANCE_EVENT].(int32)
+					if useResetQStats && evEnabled == 0 {
+						err = fmt.Errorf("Requested use of RESET QSTATS but queue manager has PERFMEV(DISABLED)")
+						errorString = "Command"
+					}
 				} else {
 					if cc.UsePublications == true {
 						if commandLevel < 900 && platform != ibmmq.MQPL_APPLIANCE {
@@ -143,6 +152,8 @@ func InitConnection(qMgrName string, replyQ string, cc *ConnectionConfig) error
 			// Don't need the qMgrObject any more
 			qMgrObject.Close(0)
 
+		} else {
+			errorString = "Cannot open queue manager object"
 		}
 	}
 
@@ -189,7 +200,7 @@ func InitConnection(qMgrName string, replyQ string, cc *ConnectionConfig) error
 		mqod.ObjectName = replyQ
 		statusReplyQObj, err = qMgr.Open(mqod, openOptions)
 		if err != nil {
-			errorString = "Cannot open queue" + replyQ
+			errorString = "Cannot open queue " + replyQ
 		}
 	}
 

mqmetric/queue.go

@@ -30,8 +30,8 @@ about MQ queues
 */
 
 import (
-	"github.com/ibm-messaging/mq-golang/ibmmq"
 	//	"fmt"
+	"github.com/ibm-messaging/mq-golang/ibmmq"
 	"strings"
 	"time"
 )
@@ -48,6 +48,17 @@ const (
 	ATTR_Q_SINCE_GET   = "time_since_get"
 	ATTR_Q_MAX_DEPTH   = "attribute_max_depth"
 	ATTR_Q_USAGE       = "attribute_usage"
+
+	// The next two attributes are given the same name
+	// as the published statistics from the amqsrua-style
+	// vaues. That allows a dashboard for Distributed and z/OS
+	// to merge the same query.
+	ATTR_Q_INTERVAL_PUT = "mqput_mqput1_count"
+	ATTR_Q_INTERVAL_GET = "mqget_count"
+	// This is the Highest Depth returned over an interval via the
+	// RESET QSTATS command. Contrast with the attribute_max_depth
+	// value which is the DISPLAY QL(x) MAXDEPTH attribute.
+	ATTR_Q_INTERVAL_HI_DEPTH = "hi_depth"
 )
 
 var QueueStatus StatusSet
@@ -89,9 +100,19 @@ func QueueInitAttributes() {
 		QueueStatus.Attributes[attr] = newStatusAttribute(attr, "Queue Depth", ibmmq.MQIA_CURRENT_Q_DEPTH)
 	}
 
+	if platform == ibmmq.MQPL_ZOS && useResetQStats {
+		attr = ATTR_Q_INTERVAL_PUT
+		QueueStatus.Attributes[attr] = newStatusAttribute(attr, "Put/Put1 Count", ibmmq.MQIA_MSG_ENQ_COUNT)
+		attr = ATTR_Q_INTERVAL_GET
+		QueueStatus.Attributes[attr] = newStatusAttribute(attr, "Get Count", ibmmq.MQIA_MSG_DEQ_COUNT)
+		attr = ATTR_Q_INTERVAL_HI_DEPTH
+		QueueStatus.Attributes[attr] = newStatusAttribute(attr, "Highest Depth", ibmmq.MQIA_HIGH_Q_DEPTH)
+	}
+
 	// This is not really a monitoring metric but it enables calculations to be made such as %full for
 	// the queue. It's extracted at startup of the program via INQUIRE_Q and not updated later even if the
-	// queue definition is changed. It's not easy to generate the % value in this program as the CurDepth will
+	// queue definition is changed until rediscovery of the queues on a schedule.
+	// It's not easy to generate the % value in this program as the CurDepth will
 	// usually - but not always - come from the published resource stats. So we don't have direct access to it.
 	// Recording the MaxDepth allows Prometheus etc to do the calculation regardless of how the CurDepth was obtained.
 	attr = ATTR_Q_MAX_DEPTH
@@ -142,6 +163,9 @@ func CollectQueueStatus(patterns string) error {
 			}
 			//fmt.Printf("Collecting qStatus for %s\n",qName)
 			err = collectQueueStatus(qName, ibmmq.MQOT_Q)
+			if err == nil && useResetQStats {
+				err = collectResetQStats(qName)
+			}
 		}
 	} else {
 		for _, pattern := range queuePatterns {
@@ -150,6 +174,9 @@ func CollectQueueStatus(patterns string) error {
 				continue
 			}
 			err = collectQueueStatus(pattern, ibmmq.MQOT_Q)
+			if err == nil && useResetQStats {
+				err = collectResetQStats(pattern)
+			}
 		}
 	}
 	return err
@@ -204,6 +231,43 @@ func collectQueueStatus(pattern string, instanceType int32) error {
 	return err
 }
 
+func collectResetQStats(pattern string) error {
+	var err error
+
+	statusClearReplyQ()
+	putmqmd, pmo, cfh, buf := statusSetCommandHeaders()
+
+	// Can allow all the other fields to default
+	cfh.Command = ibmmq.MQCMD_RESET_Q_STATS
+
+	// Add the parameters one at a time into a buffer
+	pcfparm := new(ibmmq.PCFParameter)
+	pcfparm.Type = ibmmq.MQCFT_STRING
+	pcfparm.Parameter = ibmmq.MQCA_Q_NAME
+	pcfparm.String = []string{pattern}
+	cfh.ParameterCount++
+	buf = append(buf, pcfparm.Bytes()...)
+
+	buf = append(cfh.Bytes(), buf...)
+
+	// And now put the command to the queue
+	err = cmdQObj.Put(putmqmd, pmo, buf)
+	if err != nil {
+		return err
+	}
+
+	// Now get the responses - loop until all have been received (one
+	// per queue) or we run out of time
+	for allReceived := false; !allReceived; {
+		cfh, buf, allReceived, err = statusGetReply()
+		if buf != nil {
+			parseResetQStatsData(cfh, buf)
+		}
+	}
+
+	return err
+}
+
 // Issue the INQUIRE_Q call for wildcarded queue names and
 // extract the required attributes - currently, just the
 // Maximum Queue Depth
@@ -334,16 +398,68 @@ func parseQData(instanceType int32, cfh *ibmmq.MQCFH, buf []byte) string {
 	now := time.Now()
 	QueueStatus.Attributes[ATTR_Q_SINCE_PUT].Values[key] = newStatusValueInt64(statusTimeDiff(now, lastPutDate, lastPutTime))
 	QueueStatus.Attributes[ATTR_Q_SINCE_GET].Values[key] = newStatusValueInt64(statusTimeDiff(now, lastGetDate, lastGetTime))
-
 	if s, ok := qInfoMap[key]; ok {
-		maxDepth := s.MaxDepth
+		maxDepth := s.AttrMaxDepth
 		QueueStatus.Attributes[ATTR_Q_MAX_DEPTH].Values[key] = newStatusValueInt64(maxDepth)
-		usage := s.Usage
+		usage := s.AttrUsage
 		QueueStatus.Attributes[ATTR_Q_USAGE].Values[key] = newStatusValueInt64(usage)
 	}
 	return key
 }
 
+// Given a PCF response message, parse it to extract the desired statistics
+func parseResetQStatsData(cfh *ibmmq.MQCFH, buf []byte) string {
+	var elem *ibmmq.PCFParameter
+
+	qName := ""
+	key := ""
+
+	parmAvail := true
+	bytesRead := 0
+	offset := 0
+	datalen := len(buf)
+	if cfh == nil || cfh.ParameterCount == 0 {
+		return ""
+	}
+
+	// Parse it once to extract the fields that are needed for the map key
+	for parmAvail && cfh.CompCode != ibmmq.MQCC_FAILED {
+		elem, bytesRead = ibmmq.ReadPCFParameter(buf[offset:])
+		offset += bytesRead
+		// Have we now reached the end of the message
+		if offset >= datalen {
+			parmAvail = false
+		}
+
+		// Only one field needed for queues
+		switch elem.Parameter {
+		case ibmmq.MQCA_Q_NAME:
+			qName = strings.TrimSpace(elem.String[0])
+		}
+	}
+
+	// Create a unique key for this instance
+	key = qName
+
+	QueueStatus.Attributes[ATTR_Q_NAME].Values[key] = newStatusValueString(qName)
+
+	// And then re-parse the message so we can store the metrics now knowing the map key
+	parmAvail = true
+	offset = 0
+	for parmAvail && cfh.CompCode != ibmmq.MQCC_FAILED {
+		elem, bytesRead = ibmmq.ReadPCFParameter(buf[offset:])
+		offset += bytesRead
+		// Have we now reached the end of the message
+		if offset >= datalen {
+			parmAvail = false
+		}
+
+		statusGetIntAttributes(QueueStatus, elem, key)
+	}
+
+	return key
+}
+
 func parseQAttrData(cfh *ibmmq.MQCFH, buf []byte) {
 	var elem *ibmmq.PCFParameter
 
@@ -388,15 +504,15 @@ func parseQAttrData(cfh *ibmmq.MQCFH, buf []byte) {
 			v := elem.Int64Value[0]
 			if v > 0 {
 				if qInfo, ok := qInfoMap[qName]; ok {
-					qInfo.MaxDepth = v
+					qInfo.AttrMaxDepth = v
 				}
 			}
 			//fmt.Printf("MaxQDepth for %s = %d \n",qName,v)
 		case ibmmq.MQIA_USAGE:
 			v := elem.Int64Value[0]
 			if v > 0 {
 				if qInfo, ok := qInfoMap[qName]; ok {
-					qInfo.Usage = v
+					qInfo.AttrUsage = v
 				}
 			}
 		}