QPID-945 from Ted Ross

git-svn-id: https://svn.apache.org/repos/asf/incubator/qpid/trunk/qpid@650620 13f79535-47bb-0310-9956-ffa450edef68

1 files changed, 159 insertions, 0 deletions

diff --git a/python/commands/qpid-queue-stats b/python/commands/qpid-queue-stats
new file mode 100755
index 0000000000..6e3579edfd
--- /dev/null
+++ b/python/commands/qpid-queue-stats
@@ -0,0 +1,159 @@
+#!/usr/bin/env python
+
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you 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.
+#
+
+import os
+import getopt
+import sys
+import socket
+import qpid
+import uuid
+from threading       import Condition
+from qpid.management import managementClient
+from qpid.peer       import Closed
+from qpid.connection import Connection
+from qpid.util       import connect
+from time            import sleep
+
+defspecpath  = "/usr/share/amqp/amqp.0-10.xml"
+specpath     = defspecpath
+host         = "localhost"
+
+def Usage ():
+    print "Usage:  qpid-queue-stats [OPTIONS]"
+    print
+    print "Options:"
+    print "    -a <broker-addr>     default: localhost"
+    print "         broker-addr is in the form:   hostname | ip-address [:<port>]"
+    print "         ex:  localhost, 10.1.1.7:10000, broker-host:10000"
+    print "    -s <amqp-spec-file>  default:", defspecpath
+    print
+    sys.exit (1)
+
+class Broker:
+    def __init__ (self, text):
+        colon = text.find (":")
+        if colon == -1:
+            host = text
+            self.port = 5672
+        else:
+            host = text[:colon]
+            self.port = int (text[colon+1:])
+        self.host = socket.gethostbyname (host)
+
+    def name (self):
+        return self.host + ":" + str (self.port)
+
+class mgmtObject (object):
+  """ Generic object that holds the contents of a management object with its
+      attributes set as object attributes. """
+
+  def __init__ (self, classKey, timestamps, row):
+    self.classKey   = classKey
+    self.timestamps = timestamps
+    for cell in row:
+      setattr (self, cell[0], cell[1])
+
+
+
+class BrokerManager:
+    def __init__ (self):
+        self.dest    = None
+        self.src     = None
+        self.broker  = None
+        self.objects = {}
+
+    def SetBroker (self, broker):
+        self.broker = broker
+
+    def ConnectToBroker (self):
+        try:
+            self.spec     = qpid.spec.load (specpath)
+            self.conn     = Connection (connect (self.broker.host, self.broker.port), self.spec)
+            self.conn.start ()
+            self.mclient  = managementClient (self.spec, None, self.configCb, self.instCb)
+            self.mchannel = self.mclient.addChannel (self.conn.session(str(uuid.uuid4())))
+        except socket.error, e:
+            print "Connect Error:", e
+            exit (1)
+
+    def configCb (self, context, classKey, row, timestamps):
+        className = classKey[1]
+        if className != "queue":
+            return
+
+        obj = mgmtObject (classKey, timestamps, row)
+        if obj.id not in self.objects:
+            self.objects[obj.id] = (obj.name, None, None)
+
+    def instCb (self, context, classKey, row, timestamps):
+        className = classKey[1]
+        if className != "queue":
+            return
+
+        obj = mgmtObject (classKey, timestamps, row)
+        if obj.id not in self.objects:
+            return
+
+        (name, first, last) = self.objects[obj.id]
+        if first == None:
+            self.objects[obj.id] = (name, obj, None)
+            return
+
+        if last == None:
+            lastSample = first
+        else:
+            lastSample = last
+
+        self.objects[obj.id] = (name, first, obj)
+
+        deltaTime   = float (obj.timestamps[0] - lastSample.timestamps[0])
+        enqueueRate = float (obj.msgTotalEnqueues - lastSample.msgTotalEnqueues) / (deltaTime / 1000000000.0)
+        dequeueRate = float (obj.msgTotalDequeues - lastSample.msgTotalDequeues) / (deltaTime / 1000000000.0)
+        print "%-41s%10.2f%10d..%-10d%13.2f%13.2f" % \
+            (name, deltaTime / 1000000000, obj.msgDepthLow, obj.msgDepthHigh, enqueueRate, dequeueRate)
+
+
+    def Overview (self):
+        self.ConnectToBroker ()
+        print "Queue Name                                     Sec       Depth Range          Enq Rate     Deq Rate"
+        print "==================================================================================================="
+        while True:
+            sleep (1)
+
+##
+## Main Program
+##
+
+try:
+    (optlist, cargs) = getopt.getopt (sys.argv[1:], "s:a:")
+except:
+    Usage ()
+
+for opt in optlist:
+    if opt[0] == "-s":
+        specpath = opt[1]
+    if opt[0] == "-a":
+        host = opt[1]
+
+nargs = len (cargs)
+bm  = BrokerManager ()
+bm.SetBroker (Broker (host))
+bm.Overview ()