QPID-1327 - Added severity field to events, cleaned up routing key usage

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

8 files changed, 89 insertions, 46 deletions

diff --git a/cpp/managementgen/qmf/schema.py b/cpp/managementgen/qmf/schema.py
index 48e697ab4a..7f008c19e3 100755
--- a/cpp/managementgen/qmf/schema.py
+++ b/cpp/managementgen/qmf/schema.py
@@ -706,6 +706,7 @@ class SchemaEvent:
     self.packageName = package
     self.name = None
     self.desc = None
+    self.sevText = "inform"
     self.args = []
     self.hash = Hash(node)
 
@@ -719,6 +720,9 @@ class SchemaEvent:
       elif key == 'desc':
         self.desc = val
 
+      elif key == 'sev':
+        self.sevText = val
+
       elif key == 'args':
         list = val.replace(" ", "").split(",")
         for item in list:
@@ -730,6 +734,17 @@ class SchemaEvent:
       else:
         raise ValueError ("Unknown attribute in event '%s'" % key)
 
+    if   self.sevText == "emerg"  : self.sev = 0
+    elif self.sevText == "alert"  : self.sev = 1
+    elif self.sevText == "crit"   : self.sev = 2
+    elif self.sevText == "error"  : self.sev = 3
+    elif self.sevText == "warn"   : self.sev = 4
+    elif self.sevText == "notice" : self.sev = 5
+    elif self.sevText == "inform" : self.sev = 6
+    elif self.sevText == "debug"  : self.sev = 7
+    else:
+      raise ValueError("Unknown severity '%s' in event '%s'" % (self.sevText, self.name))
+
   def getName (self):
     return self.name
 
@@ -796,6 +811,7 @@ class SchemaEvent:
       stream.write ("namespace %s {\n" % item)
 
   def genArgEncodes(self, stream, variables):
+    stream.write("    buf.putOctet(%d);\n" % self.sev)
     for arg in self.args:
       stream.write("    " + arg.type.type.encode.replace("@", "buf").replace("#", arg.name) + ";\n")
 
diff --git a/cpp/src/qpid/acl/management-schema.xml b/cpp/src/qpid/acl/management-schema.xml
index f362561356..661228b4d3 100644
--- a/cpp/src/qpid/acl/management-schema.xml
+++ b/cpp/src/qpid/acl/management-schema.xml
@@ -36,9 +36,9 @@
     <arg name="userId"     type="sstr"/>
   </eventArguments>
 
-  <event name="allow"          args="userId, action, objectType, objectName, arguments"/>
-  <event name="deny"           args="userId, action, objectType, objectName, arguments"/>
-  <event name="fileLoaded"     args="userId"/>
-  <event name="fileLoadFailed" args="userId, reason"/>
+  <event name="allow"          sev="inform" args="userId, action, objectType, objectName, arguments"/>
+  <event name="deny"           sev="notice" args="userId, action, objectType, objectName, arguments"/>
+  <event name="fileLoaded"     sev="inform" args="userId"/>
+  <event name="fileLoadFailed" sev="error"  args="userId, reason"/>
 
 </schema>
diff --git a/cpp/src/qpid/agent/ManagementAgentImpl.cpp b/cpp/src/qpid/agent/ManagementAgentImpl.cpp
index b178d0fc28..06b61b819a 100644
--- a/cpp/src/qpid/agent/ManagementAgentImpl.cpp
+++ b/cpp/src/qpid/agent/ManagementAgentImpl.cpp
@@ -172,7 +172,8 @@ void ManagementAgentImpl::raiseEvent(const ManagementEvent& event)
     event.encode(outBuffer);
     outLen = MA_BUFFER_SIZE - outBuffer.available();
     outBuffer.reset();
-    connThreadBody.sendBuffer(outBuffer, outLen, "qpid.management", "mgmt.event");
+    connThreadBody.sendBuffer(outBuffer, outLen, "qpid.management",
+                              "console.event." + event.getPackageName() + "." + event.getEventName());
 }
 
 uint32_t ManagementAgentImpl::pollCallbacks(uint32_t callLimit)
@@ -510,7 +511,7 @@ ManagementAgentImpl::PackageMap::iterator ManagementAgentImpl::findOrAddPackage(
     encodePackageIndication(outBuffer, result.first);
     outLen = MA_BUFFER_SIZE - outBuffer.available();
     outBuffer.reset();
-    connThreadBody.sendBuffer(outBuffer, outLen, "qpid.management", "mgmt.schema.package");
+    connThreadBody.sendBuffer(outBuffer, outLen, "qpid.management", "schema.package");
 
     return result.first;
 }
@@ -582,7 +583,7 @@ void ManagementAgentImpl::periodicProcessing()
 
         contentSize = BUFSIZE - msgBuffer.available();
         msgBuffer.reset();
-        routingKey = "mgmt." + systemId.str() + ".heartbeat";
+        routingKey = "console.heartbeat";
         connThreadBody.sendBuffer(msgBuffer, contentSize, "qpid.management", routingKey);
     }
 
@@ -617,7 +618,7 @@ void ManagementAgentImpl::periodicProcessing()
 
             contentSize = BUFSIZE - msgBuffer.available();
             msgBuffer.reset();
-            routingKey = "mgmt." + systemId.str() + ".prop." + object->getClassName();
+            routingKey = "console.prop." + object->getPackageName() + "." + object->getClassName();
             connThreadBody.sendBuffer(msgBuffer, contentSize, "qpid.management", routingKey);
         }
         
@@ -629,7 +630,7 @@ void ManagementAgentImpl::periodicProcessing()
 
             contentSize = BUFSIZE - msgBuffer.available();
             msgBuffer.reset();
-            routingKey = "mgmt." + systemId.str() + ".stat." + object->getClassName();
+            routingKey = "console.stat." + object->getPackageName() + "." + object->getClassName();
             connThreadBody.sendBuffer(msgBuffer, contentSize, "qpid.management", routingKey);
         }
 
diff --git a/cpp/src/qpid/management/ManagementBroker.cpp b/cpp/src/qpid/management/ManagementBroker.cpp
index 0e046bb813..e1770f0084 100644
--- a/cpp/src/qpid/management/ManagementBroker.cpp
+++ b/cpp/src/qpid/management/ManagementBroker.cpp
@@ -235,7 +235,8 @@ void ManagementBroker::raiseEvent(const ManagementEvent& event)
     event.encode(outBuffer);
     outLen = MA_BUFFER_SIZE - outBuffer.available();
     outBuffer.reset();
-    sendBuffer(outBuffer, outLen, mExchange, "mgmt.event");
+    sendBuffer(outBuffer, outLen, mExchange,
+               "console.event." + event.getPackageName() + "." + event.getEventName());
 }
 
 ManagementBroker::Periodic::Periodic (ManagementBroker& _broker, uint32_t _seconds)
@@ -347,7 +348,7 @@ void ManagementBroker::periodicProcessing (void)
 
         contentSize = BUFSIZE - msgBuffer.available ();
         msgBuffer.reset ();
-        routingKey = "mgmt." + uuid.str() + ".heartbeat";
+        routingKey = "console.heartbeat";
         sendBuffer (msgBuffer, contentSize, mExchange, routingKey);
     }
 
@@ -382,7 +383,7 @@ void ManagementBroker::periodicProcessing (void)
 
             contentSize = BUFSIZE - msgBuffer.available ();
             msgBuffer.reset ();
-            routingKey = "mgmt." + uuid.str() + ".prop." + object->getClassName ();
+            routingKey = "console.prop." + object->getPackageName() + "." + object->getClassName ();
             sendBuffer (msgBuffer, contentSize, mExchange, routingKey);
         }
         
@@ -394,7 +395,7 @@ void ManagementBroker::periodicProcessing (void)
 
             contentSize = BUFSIZE - msgBuffer.available ();
             msgBuffer.reset ();
-            routingKey = "mgmt." + uuid.str () + ".stat." + object->getClassName ();
+            routingKey = "console.stat." + object->getPackageName() + "." + object->getClassName ();
             sendBuffer (msgBuffer, contentSize, mExchange, routingKey);
         }
 
@@ -440,6 +441,7 @@ bool ManagementBroker::dispatchCommand (Deliverable&      deliverable,
     //
     //    agent.1.0.#
     //    broker
+    //    schema.#
 
     if (routingKey == "broker") {
         dispatchAgentCommandLH(msg);
@@ -455,6 +457,11 @@ bool ManagementBroker::dispatchCommand (Deliverable&      deliverable,
         return authorizeAgentMessageLH(msg);
     }
 
+    else if (routingKey.compare(0, 7, "schema.") == 0) {
+        dispatchAgentCommandLH(msg);
+        return true;
+    }
+
     return true;
 }
 
@@ -698,7 +705,7 @@ void ManagementBroker::handleSchemaResponseLH(Buffer& inBuffer, string /*replyTo
                 encodeClassIndication(outBuffer, pIter, cIter);
                 outLen = MA_BUFFER_SIZE - outBuffer.available();
                 outBuffer.reset();
-                sendBuffer(outBuffer, outLen, mExchange, "mgmt." + uuid.str() + ".schema");
+                sendBuffer(outBuffer, outLen, mExchange, "schema.class");
             }
         }
     }
@@ -972,7 +979,7 @@ ManagementBroker::PackageMap::iterator ManagementBroker::findOrAddPackageLH(std:
     encodePackageIndication (outBuffer, result.first);
     outLen = MA_BUFFER_SIZE - outBuffer.available ();
     outBuffer.reset ();
-    sendBuffer (outBuffer, outLen, mExchange, "mgmt." + uuid.str() + ".schema.package");
+    sendBuffer (outBuffer, outLen, mExchange, "schema.package");
 
     return result.first;
 }
diff --git a/java/client/src/main/java/org/apache/qpid/management/Names.java b/java/client/src/main/java/org/apache/qpid/management/Names.java
index 64e5c0fef2..dde7715509 100644
--- a/java/client/src/main/java/org/apache/qpid/management/Names.java
+++ b/java/client/src/main/java/org/apache/qpid/management/Names.java
@@ -29,7 +29,7 @@ public interface Names
 {
     /** Name of the qpid management exchange. */
     String MANAGEMENT_EXCHANGE = "qpid.management";    
-    String MANAGEMENT_ROUTING_KEY = "mgmt.#";
+    String MANAGEMENT_ROUTING_KEY = "console.#";
    
     String MANAGEMENT_QUEUE_PREFIX = "management.";
     String METHOD_REPLY_QUEUE_PREFIX = "reply.";
diff --git a/python/qpid/management.py b/python/qpid/management.py
index 81d9dbe030..e6ff5852f2 100644
--- a/python/qpid/management.py
+++ b/python/qpid/management.py
@@ -522,7 +522,9 @@ class managementClient:
       if self.ctrlCb != None:
         self.ctrlCb (ch.context, self.CTRL_SCHEMA_LOADED, None)
       ch.ssn.exchange_bind (exchange="qpid.management",
-                            queue=ch.topicName, binding_key="mgmt.#")
+                            queue=ch.topicName, binding_key="console.#")
+      ch.ssn.exchange_bind (exchange="qpid.management",
+                            queue=ch.topicName, binding_key="schema.#")
 
 
   def handleMethodReply (self, ch, codec, sequence):
diff --git a/python/qpid/qmfconsole.py b/python/qpid/qmfconsole.py
index c8035e87f2..bcd1e59819 100644
--- a/python/qpid/qmfconsole.py
+++ b/python/qpid/qmfconsole.py
@@ -141,6 +141,10 @@ class Session:
     self.getResult         = []
     self.getSelect         = []
     self.error             = None
+    if self.console == None:
+      rcvObjects    = False
+      rcvEvents     = False
+      rcvHeartbeats = False
     self.bindingKeyList    = self._bindingKeys(rcvObjects, rcvEvents, rcvHeartbeats)
     self.manageConnections = manageConnections
 
@@ -316,16 +320,19 @@ class Session:
 
   def _bindingKeys(self, rcvObjects, rcvEvents, rcvHeartbeats):
     keyList = []
+    keyList.append("schema.#")
     if rcvObjects and rcvEvents and rcvHeartbeats:
-      keyList.append("mgmt.#")
+      keyList.append("console.#")
     else:
       if rcvObjects:
-        keyList.append("mgmt.*.prop.#")
-        keyList.append("mgmt.*.stat.#")
+        keyList.append("console.prop.#")
+        keyList.append("console.stat.#")
+      else:
+        keyList.append("console.prop.org.apache.qpid.broker.agent")
       if rcvEvents:
-        keyList.append("mgmt.event")
+        keyList.append("console.event.#")
       if rcvHeartbeats:
-        keyList.append("mgmt.*.heartbeat.#")
+        keyList.append("console.heartbeat")
     return keyList
 
   def _handleBrokerConnect(self, broker):
@@ -1013,14 +1020,13 @@ class Broker:
       self.amqpSession.message_flow(destination="rdest", unit=0, value=0xFFFFFFFF)
       self.amqpSession.message_flow(destination="rdest", unit=1, value=0xFFFFFFFF)
 
-      if self.session.console != None:
-        self.topicName = "topic-%s" % self.amqpSessionId
-        self.amqpSession.queue_declare(queue=self.topicName, exclusive=True, auto_delete=True)
-        self.amqpSession.message_subscribe(queue=self.topicName, destination="tdest")
-        self.amqpSession.incoming("tdest").listen(self._replyCb)
-        self.amqpSession.message_set_flow_mode(destination="tdest", flow_mode=1)
-        self.amqpSession.message_flow(destination="tdest", unit=0, value=0xFFFFFFFF)
-        self.amqpSession.message_flow(destination="tdest", unit=1, value=0xFFFFFFFF)
+      self.topicName = "topic-%s" % self.amqpSessionId
+      self.amqpSession.queue_declare(queue=self.topicName, exclusive=True, auto_delete=True)
+      self.amqpSession.message_subscribe(queue=self.topicName, destination="tdest")
+      self.amqpSession.incoming("tdest").listen(self._replyCb)
+      self.amqpSession.message_set_flow_mode(destination="tdest", flow_mode=1)
+      self.amqpSession.message_flow(destination="tdest", unit=0, value=0xFFFFFFFF)
+      self.amqpSession.message_flow(destination="tdest", unit=1, value=0xFFFFFFFF)
 
       self.isConnected = True
       self.session._handleBrokerConnect(self)
@@ -1116,7 +1122,7 @@ class Broker:
   def _decOutstanding(self):
     self.cv.acquire()
     self.reqsOutstanding -= 1
-    if self.reqsOutstanding == 0 and not self.topicBound and self.session.console != None:
+    if self.reqsOutstanding == 0 and not self.topicBound:
       self.topicBound = True
       for key in self.session.bindingKeyList:
         self.amqpSession.exchange_bind(exchange="qpid.management",
@@ -1174,6 +1180,7 @@ class Event:
     hash  = codec.read_bin128()
     self.classKey = (pname, cname, hash)
     self.timestamp = codec.read_int64()
+    self.severity = codec.read_uint8()
     self.schema = None
     if pname in session.packages:
       if (cname, hash) in session.packages[pname]:
@@ -1185,6 +1192,16 @@ class Event:
   def __repr__(self):
     return self.getSyslogText()
 
+  def _sevName(self):
+    if self.severity == 0 : return "EMER "
+    if self.severity == 1 : return "ALERT"
+    if self.severity == 2 : return "CRIT "
+    if self.severity == 3 : return "ERROR"
+    if self.severity == 4 : return "WARN "
+    if self.severity == 5 : return "NOTIC"
+    if self.severity == 6 : return "INFO "
+    if self.severity == 7 : return "DEBUG"
+
   def getClassKey(self):
     return self.classKey
 
@@ -1204,7 +1221,7 @@ class Event:
     if self.schema == None:
       return "<uninterpretable>"
     out = strftime("%c", gmtime(self.timestamp / 1000000000))
-    out += " " + self.classKey[0] + ":" + self.classKey[1]
+    out += " " + self._sevName() + " " + self.classKey[0] + ":" + self.classKey[1]
     for arg in self.schema.arguments:
       out += " " + arg.name + "=" + self.session._displayValue(self.arguments[arg.name], arg.type)
     return out
diff --git a/specs/management-schema.xml b/specs/management-schema.xml
index dc42a4108e..7a0543677f 100644
--- a/specs/management-schema.xml
+++ b/specs/management-schema.xml
@@ -312,19 +312,19 @@
     <arg name="user"    type="sstr"   desc="Authentication identity"/>
   </eventArguments>
 
-  <event name="clientConnect"     args="rhost, user"/>
-  <event name="clientDisconnect"  args="rhost, user"/>
-  <event name="agentConnect"      args="rhost, user"/>
-  <event name="agentDisconnect"   args="rhost, user"/>
-  <event name="brokerConnect"     args="rhost, user"/>
-  <event name="brokerDisconnect"  args="rhost, user"/>
-  <event name="queueDeclare"      args="rhost, user, qName, durable, excl, autoDel, args, disp"/>
-  <event name="queueDelete"       args="rhost, user, qName"/>
-  <event name="exchangeDeclare"   args="rhost, user, exName, exType, altEx, durable, autoDel, args, disp"/>
-  <event name="exchangeDelete"    args="rhost, user, exName"/>
-  <event name="bind"              args="rhost, user, exName, qName, key, args"/>
-  <event name="unbind"            args="rhost, user, exName, qName, key"/>
-  <event name="subscribe"         args="rhost, user, qName, dest, excl, args"/>
-  <event name="unsubscribe"       args="rhost, user, dest"/>
+  <event name="clientConnect"     sev="inform" args="rhost, user"/>
+  <event name="clientDisconnect"  sev="inform" args="rhost, user"/>
+  <event name="agentConnect"      sev="inform" args="rhost, user"/>
+  <event name="agentDisconnect"   sev="inform" args="rhost, user"/>
+  <event name="brokerConnect"     sev="inform" args="rhost, user"/>
+  <event name="brokerDisconnect"  sev="inform" args="rhost, user"/>
+  <event name="queueDeclare"      sev="inform" args="rhost, user, qName, durable, excl, autoDel, args, disp"/>
+  <event name="queueDelete"       sev="inform" args="rhost, user, qName"/>
+  <event name="exchangeDeclare"   sev="inform" args="rhost, user, exName, exType, altEx, durable, autoDel, args, disp"/>
+  <event name="exchangeDelete"    sev="inform" args="rhost, user, exName"/>
+  <event name="bind"              sev="inform" args="rhost, user, exName, qName, key, args"/>
+  <event name="unbind"            sev="inform" args="rhost, user, exName, qName, key"/>
+  <event name="subscribe"         sev="inform" args="rhost, user, qName, dest, excl, args"/>
+  <event name="unsubscribe"       sev="inform" args="rhost, user, dest"/>
 </schema>