1 files changed, 190 insertions, 66 deletions

diff --git a/ndb/src/mgmsrv/Services.cpp b/ndb/src/mgmsrv/Services.cpp
index 7bf408583de..5b552836955 100644
--- a/ndb/src/mgmsrv/Services.cpp
+++ b/ndb/src/mgmsrv/Services.cpp
@@ -23,6 +23,7 @@
 #include <mgmapi.h>
 #include <EventLogger.hpp>
 #include <signaldata/SetLogLevelOrd.hpp>
+#include <LogLevel.hpp>
 #include <BaseString.hpp>
 #include <Base64.hpp>
 
@@ -133,7 +134,7 @@ ParserRow<MgmApiSession> commands[] = {
     MGM_ARG("public key", String, Mandatory, "Public key"),
 
   MGM_CMD("get version", &MgmApiSession::getVersion, ""),
-
+  
   MGM_CMD("get status", &MgmApiSession::getStatus, ""),
 
   MGM_CMD("get info clusterlog", &MgmApiSession::getInfoClusterLog, ""),
@@ -236,7 +237,11 @@ ParserRow<MgmApiSession> commands[] = {
     MGM_ARG("node", String, Mandatory, "Node"),
     MGM_ARG("parameter", String, Mandatory, "Parameter"),
     MGM_ARG("value", String, Mandatory, "Value"),
-  
+
+  MGM_CMD("listen event", &MgmApiSession::listen_event, ""),
+    MGM_ARG("node", Int, Optional, "Node"),  
+    MGM_ARG("filter", String, Mandatory, "Event category"),
+
   MGM_END()
 };
 
@@ -289,7 +294,8 @@ MgmApiSession::runSession() {
       break;
     }
   }
-  NDB_CLOSE_SOCKET(m_socket);
+  if(m_socket >= 0)
+    NDB_CLOSE_SOCKET(m_socket);
 }
 
 #ifdef MGM_GET_CONFIG_BACKWARDS_COMPAT
@@ -413,11 +419,15 @@ MgmApiSession::get_nodeid(Parser_t::Context &,
 
   NodeId tmp= nodeid;
   if(tmp == 0 || !m_allocated_resources->is_reserved(tmp)){
+    BaseString error_string;
     if (!m_mgmsrv.alloc_node_id(&tmp, (enum ndb_mgm_node_type)nodetype, 
-				&addr, &addrlen)){
+				&addr, &addrlen, error_string)){
+      const char *alias;
+      const char *str;
+      alias= ndb_mgm_get_node_type_alias_string((enum ndb_mgm_node_type)
+						nodetype, &str);
       m_output->println(cmd);
-      m_output->println("result: no free nodeid %d for nodetype %d",
-			nodeid, nodetype);
+      m_output->println("result: %s", error_string.c_str());
       m_output->println("");
       return;
     }
@@ -551,7 +561,7 @@ MgmApiSession::getStatPort(Parser_t::Context &,
 			   const class Properties &) {
 
   m_output->println("get statport reply");
-  m_output->println("tcpport: %d", m_mgmsrv.getStatPort());
+  m_output->println("tcpport: %d", 0);
   m_output->println("");
 }
 
@@ -753,13 +763,12 @@ MgmApiSession::bye(Parser<MgmApiSession>::Context &,
 void
 MgmApiSession::setClusterLogLevel(Parser<MgmApiSession>::Context &,
 				  Properties const &args) {
-  Uint32 node, level;
-  BaseString categoryName, errorString;
+  Uint32 node, level, category;
+  BaseString errorString;
   SetLogLevelOrd logLevel;
   int result;
-  logLevel.clear();
   args.get("node", &node);
-  args.get("category", categoryName);
+  args.get("category", &category);
   args.get("level", &level);
 
   /* XXX should use constants for this value */
@@ -768,25 +777,17 @@ MgmApiSession::setClusterLogLevel(Parser<MgmApiSession>::Context &,
     goto error;
   }
 
-  categoryName.ndb_toupper();
- 
-  LogLevel::EventCategory category;
-  if(!EventLogger::matchEventCategory(categoryName.c_str(), &category)) {
-    errorString.assign("Unknown category");
-    goto error;
-  }
-
-  logLevel.setLogLevel(category, level);
-  result = m_mgmsrv.setEventReportingLevel(node, logLevel);
-
+  EventSubscribeReq req;
+  req.blockRef = 0;
+  req.noOfEntries = 1;
+  req.theData[0] = (category << 16) | level;
+  m_mgmsrv.m_log_level_requests.push_back(req);
+  
   m_output->println("set cluster loglevel reply");
-  if(result != 0)
-    m_output->println("result: %s", m_mgmsrv.getErrorText(result));
-  else
-    m_output->println("result: Ok");
+  m_output->println("result: Ok");
   m_output->println("");
   return;
- error:
+error:
   m_output->println("set cluster loglevel reply");
   m_output->println("result: %s", errorString.c_str());
   m_output->println("");
@@ -795,13 +796,13 @@ MgmApiSession::setClusterLogLevel(Parser<MgmApiSession>::Context &,
 void
 MgmApiSession::setLogLevel(Parser<MgmApiSession>::Context &,
 			   Properties const &args) {
-  Uint32 node = 0, level = 0;
-  BaseString categoryName, errorString;
+  Uint32 node = 0, level = 0, category;
+  BaseString errorString;
   SetLogLevelOrd logLevel;
   int result;
   logLevel.clear();
   args.get("node", &node);
-  args.get("category", categoryName);
+  args.get("category", &category);
   args.get("level", &level);
 
   /* XXX should use constants for this value */
@@ -810,23 +811,14 @@ MgmApiSession::setLogLevel(Parser<MgmApiSession>::Context &,
     goto error;
   }
 
-  categoryName.ndb_toupper();
- 
-  LogLevel::EventCategory category;
-  if(!EventLogger::matchEventCategory(categoryName.c_str(), &category)) {
-    errorString.assign("Unknown category");
-    goto error;
-  }
-
-  logLevel.setLogLevel(category, level);
-
-  result = m_mgmsrv.setNodeLogLevel(node, logLevel);
-
+  EventSubscribeReq req;
+  req.blockRef = node;
+  req.noOfEntries = 1;
+  req.theData[0] = (category << 16) | level;
+  m_mgmsrv.m_log_level_requests.push_back(req);
+  
   m_output->println("set loglevel reply");
-  if(result != 0)
-    m_output->println("result: %s", m_mgmsrv.getErrorText(result));
-  else 
-    m_output->println("result: Ok");
+  m_output->println("result: Ok");
   m_output->println("");
   return;
  error:
@@ -943,6 +935,7 @@ printNodeStatus(OutputStream *output,
     output->println("node.%d.dynamic_id: %d", nodeId, dynamicId);
     output->println("node.%d.node_group: %d", nodeId, nodeGroup);
     output->println("node.%d.connect_count: %d", nodeId, connectCount);
+    output->println("node.%d.address: %s", nodeId, mgmsrv.get_connect_address(nodeId));
   }
 
 }
@@ -1015,8 +1008,9 @@ MgmApiSession::stop(Parser<MgmApiSession>::Context &,
   }
 
   int stop_self= 0;
+  size_t i;
 
-  for(size_t i=0; i < nodes.size(); i++) {
+  for(i=0; i < nodes.size(); i++) {
     if (nodes[i] == m_mgmsrv.getOwnNodeId()) {
       stop_self= 1;
       if (i != nodes.size()-1) {
@@ -1030,13 +1024,13 @@ MgmApiSession::stop(Parser<MgmApiSession>::Context &,
 
   int stopped = 0, result = 0;
   
-  for(size_t i=0; i < nodes.size(); i++)
+  for(i=0; i < nodes.size(); i++)
     if (nodes[i] != m_mgmsrv.getOwnNodeId()) {
       if((result = m_mgmsrv.stopNode(nodes[i], abort != 0)) == 0)
 	stopped++;
     } else
       stopped++;
-  
+
   m_output->println("stop reply");
   if(result != 0)
     m_output->println("result: %s", m_mgmsrv.getErrorText(result));
@@ -1129,7 +1123,7 @@ MgmApiSession::logSignals(Parser<MgmApiSession>::Context &,
   args.get("blocks", blockList);
   // fast fix - pekka
   char buf[200];
-  snprintf(buf, 200, "%s", blockList.c_str());
+  BaseString::snprintf(buf, 200, "%s", blockList.c_str());
   Vector<BaseString> blocks;  
 
   blockName=strtok(buf,"|");
@@ -1244,33 +1238,94 @@ MgmApiSession::configChange(Parser_t::Context &,
   m_output->println("");
 }
 
-void 
-MgmStatService::println_statistics(const BaseString &line){
-  MutexVector<NDB_SOCKET_TYPE> copy(m_sockets.size()); 
-  m_sockets.lock();
+static NdbOut&
+operator<<(NdbOut& out, const LogLevel & ll)
+{
+  out << "[LogLevel: ";
+  for(size_t i = 0; i<LogLevel::LOGLEVEL_CATEGORIES; i++)
+    out << ll.getLogLevel((LogLevel::EventCategory)i) << " ";
+  out << "]";
+  return out;
+}
+
+void
+MgmStatService::log(int eventType, const Uint32* theData, NodeId nodeId){
+  
+  Uint32 threshold = 0;
+  LogLevel::EventCategory cat= LogLevel::llInvalid;
   int i;
-  for(i = m_sockets.size() - 1; i >= 0; i--){
-    if(println_socket(m_sockets[i], MAX_WRITE_TIMEOUT, line.c_str()) == -1){
-      copy.push_back(m_sockets[i]);
-      m_sockets.erase(i, false);
+
+  for(i = 0; (unsigned)i<EventLogger::matrixSize; i++){
+    if(EventLogger::matrix[i].eventType == eventType){
+      cat = EventLogger::matrix[i].eventCategory;
+      threshold = EventLogger::matrix[i].threshold;
+      break;
+    }
+  }
+  if (cat == LogLevel::llInvalid)
+    return;
+
+  char m_text[256];
+  EventLogger::getText(m_text, sizeof(m_text), eventType, theData, nodeId);
+
+  Vector<NDB_SOCKET_TYPE> copy; 
+  m_clients.lock();
+  for(i = m_clients.size() - 1; i >= 0; i--){
+    if(threshold <= m_clients[i].m_logLevel.getLogLevel(cat)){
+      if(m_clients[i].m_socket >= 0 &&
+	 println_socket(m_clients[i].m_socket, 
+			MAX_WRITE_TIMEOUT, m_text) == -1){
+	copy.push_back(m_clients[i].m_socket);
+	m_clients.erase(i, false);
+      }
     }
   }
-  m_sockets.unlock();
+  m_clients.unlock();
   
-  for(i = copy.size() - 1; i >= 0; i--){
+  for(i = 0; (unsigned)i < copy.size(); i++){
     NDB_CLOSE_SOCKET(copy[i]);
-    copy.erase(i);
   }
-  if(m_sockets.size() == 0 || false){
-    m_mgmsrv->startStatisticEventReporting(0);
+
+  if(copy.size()){
+    LogLevel tmp; tmp.clear();
+    m_clients.lock();
+    for(i = 0; (unsigned)i < m_clients.size(); i++){
+      tmp.set_max(m_clients[i].m_logLevel);
+    }
+    m_clients.unlock();
+
+    if(!(tmp == m_logLevel)){
+      m_logLevel = tmp;
+      EventSubscribeReq req; 
+      req = tmp;
+      req.blockRef = 0;
+      m_mgmsrv->m_log_level_requests.push_back(req);
+    }
+  }
+}
+
+void
+MgmStatService::add_listener(const StatListener& client){
+  m_clients.push_back(client);
+  LogLevel tmp = m_logLevel;
+  tmp.set_max(client.m_logLevel);
+  
+  if(!(tmp == m_logLevel)){
+    m_logLevel = tmp;
+    EventSubscribeReq req;
+    req = tmp;
+    req.blockRef = 0;
+    m_mgmsrv->m_log_level_requests.push_back(req);
   }
 }
 
 void
 MgmStatService::stopSessions(){
-  for(int i = m_sockets.size() - 1; i >= 0; i--){
-    NDB_CLOSE_SOCKET(m_sockets[i]);
-    m_sockets.erase(i);
+  for(int i = m_clients.size() - 1; i >= 0; i--){
+    if(m_clients[i].m_socket >= 0){
+      NDB_CLOSE_SOCKET(m_clients[i].m_socket);
+      m_clients.erase(i);
+    }
   }
 }
 
@@ -1294,6 +1349,75 @@ MgmApiSession::setParameter(Parser_t::Context &,
   m_output->println("");
 }
 
+void
+MgmApiSession::listen_event(Parser<MgmApiSession>::Context & ctx,
+			    Properties const & args) {
+  
+  BaseString node, param, value;
+  args.get("node", node);
+  args.get("filter", param);
+
+  int result = 0;
+  BaseString msg;
+
+  MgmStatService::StatListener le;
+  le.m_socket = m_socket;
+
+  Vector<BaseString> list;
+  param.trim();
+  param.split(list, " ,");
+  for(size_t i = 0; i<list.size(); i++){
+    Vector<BaseString> spec;
+    list[i].trim();
+    list[i].split(spec, "=:");
+    if(spec.size() != 2){
+      msg.appfmt("Invalid filter specification: >%s< >%s< %d", 
+		 param.c_str(), list[i].c_str(), spec.size());
+      result = -1;
+      goto done;
+    }
+
+    spec[0].trim().ndb_toupper();
+    int category = ndb_mgm_match_event_category(spec[0].c_str());
+    if(category == NDB_MGM_ILLEGAL_EVENT_CATEGORY){
+      category = atoi(spec[0].c_str());
+      if(category < NDB_MGM_MIN_EVENT_CATEGORY ||
+	 category > NDB_MGM_MAX_EVENT_CATEGORY){
+	msg.appfmt("Unknown category: >%s<", spec[0].c_str());
+	result = -1;
+	goto done;
+      }
+    }
+    
+    int level = atoi(spec[1].c_str());
+    if(level < 0 || level > 15){
+      msg.appfmt("Invalid level: >%s<", spec[1].c_str());
+      result = -1;
+      goto done;
+    }
+    category -= CFG_MIN_LOGLEVEL;
+    le.m_logLevel.setLogLevel((LogLevel::EventCategory)category, level);
+  }
+  
+  if(list.size() == 0){
+    msg.appfmt("Empty filter specification");
+    result = -1;
+    goto done;
+  }
+  
+  m_mgmsrv.m_statisticsListner.add_listener(le);
+  
+  m_stop = true;
+  m_socket = -1;
+
+done:
+  m_output->println("listen event");
+  m_output->println("result: %d", result);
+  if(result != 0)
+    m_output->println("msg: %s", msg.c_str());
+  m_output->println("");
+}
+
 template class MutexVector<int>;
 template class Vector<ParserRow<MgmApiSession> const*>;
 template class Vector<unsigned short>;