Tutorials/i1Net/mgen: mgen_oml.2.patch

include/CWriteOml.h

@@ +1 @@
+#ifndef _CWRITEOML_H_
+#define _CWRITEOML_H_
+
+#include <iostream>
+#include <vector>
+#include <map>
+#include <string.h>
+#include "oml2/omlc.h"
+
+#include <boost/lexical_cast.hpp>
+
+
+#define FALSE (unsigned int) 0
+#define TRUE  (unsigned int) 1
+
+
+class CWriteOml
+{
+  private:
+    unsigned int bReady;
+    unsigned int _MeasurementPoints;
+    std::string mHostName;
+    OmlValueU* _values;
+
+    std::map<std::string, std::pair<OmlValueT, OmlValueU*> >            _KTVMap;
+    std::map<std::string, std::pair<OmlValueT, OmlValueU*> >::iterator  _KTVMapIter;
+
+
+    OmlMPDef* mp_def;
+    OmlMP* _mp_handle;
+    std::string _db_filename;
+    std::string _server_name;
+
+    void createMeasurementPoint(OmlMPDef* pOmlMPDef, std::string str, OmlValueT type);
+
+  public:
+    CWriteOml();
+    ~CWriteOml();
+
+    void CWriteOML(std::string db_filename, std::string server_name);
+    void init(std::string db_filename, std::string server_name);
+    void start( std::vector< std::pair<std::string, OmlValueT> >& oml_key_list );
+    void set_key(std::string key_str, void* val_ptr);
+    void insert();
+    void stop();
+
+    unsigned int isReady() { return bReady; }
+
+};
+
+#endif

include/mgen.h

@@ -5 +5 @@
 #include "mgenFlow.h"
 #include "mgenGlobals.h"
 #include "mgenMsg.h"
+#include "CWriteOml.h"
+
+
+#define DBG_OUT(x) std::cerr << "[" << __FILE__ << "] " << #x << " = " << x << std::endl;
 
 class MgenController
 {
@@ -104 +108 @@
 class Mgen 
 {
   public:
+   CWriteOml oml;
+
     enum {SCRIPT_LINE_MAX = 8192};  // maximum script line length
     
     Mgen(ProtoTimerMgr&         timerMgr, 

makefiles/Makefile.common

@@ -41 +41 @@
            $(COMMON)/mgenTransport.cpp $(COMMON)/mgenPattern.cpp \
            $(COMMON)/mgenPayload.cpp \
            $(COMMON)/mgenSequencer.cpp \
+           $(COMMON)/CWriteOml.cpp \
            $(COMMON)/gpsPub.cpp $(COMMON)/mgenAppSinkTransport.cpp
           
 MGEN_OBJ = $(MGEN_SRC:.cpp=.o)

makefiles/Makefile.linux

@@ -7 +7 @@
 #
 SYSTEM_INCLUDES = -I/usr/X11R6/include 
 SYSTEM_LDFLAGS = -L/usr/X11R6/lib 
-SYSTEM_LIBS = -ldl -lpthread -lpcap
+SYSTEM_LIBS = -ldl -lpthread -lpcap -loml2 -locomm
 
 # 2) System specific capabilities
 # Must choose appropriate for the following:

src/common/CWriteOml.cpp

@@ +1 @@
+
+#include "CWriteOml.h"
+#include <iostream>
+#include <vector>
+#include <map>
+#include <string.h>
+#include "oml2/omlc.h"
+
+//#include <boost/lexical_cast.hpp>
+
+
+CWriteOml::CWriteOml()
+{
+}
+
+void CWriteOml::CWriteOML(std::string db_filename, std::string server_name)
+{
+  bReady = FALSE;
+  init(db_filename, server_name);
+}    
+
+CWriteOml::~CWriteOml()
+{
+  omlc_close();
+}
+
+
+void CWriteOml::init(std::string db_filename, std::string server_name)
+{
+  _db_filename = db_filename;
+  _server_name = server_name;
+
+  std::string fname;
+  int argc;
+  const char** argv;
+  std::vector<char*> arg_vector;
+
+  //char chostname[32];
+  //for (int i = 0; i < 32;++i)
+  //  chostname[i] = '\0';
+  //gethostname(chostname, 31);
+  //mHostName = std::string(chostname);
+
+  std::string mode(server_name.c_str());
+
+  if (mode == "file")
+ {
+   fname = db_filename;//  + "_" +  mHostName;
+   std::cout << fname << std::endl;
+   argc = 7;
+   const char* argv_file[] = {"./spectrum", "--oml-id",(const char*)"et", "--oml-exp-id",db_filename.c_str(), "--oml-file",fname.c_str()};
+   argv = argv_file;
+  }
+  else
+  {
+    // Following two lines were being optimized out
+    //const char* argv_server[] = {"./spectrum", "--oml-id",(const char*)"et", "--oml-domain",db_filename.c_str(), "--oml-collect", server_name.c_str() };
+    //argv = argv_server;
+
+    arg_vector.push_back((char*)"./spectrum");
+    arg_vector.push_back((char*)"--oml-id");
+    arg_vector.push_back((char*)"et");
+    arg_vector.push_back((char*)"--oml-domain");
+    arg_vector.push_back((char*)db_filename.c_str());
+    arg_vector.push_back((char*)"--oml-collect");
+    arg_vector.push_back((char*)server_name.c_str());
+    argv = (const char**)&arg_vector[0];
+
+    argc = arg_vector.size(); // argc = 7;
+
+  }
+
+  int result = omlc_init ("_mp_", &argc, argv, NULL);
+  if (result == -1) {
+    std::cerr << "Could not initialize OML\n";
+    exit (1);
+  }
+
+}
+
+
+void CWriteOml::start( std::vector<std::pair<std::string, OmlValueT> >& _OmlKeys )
+{
+  int result;
+
+  _MeasurementPoints = _OmlKeys.size();
+
+  mp_def = new OmlMPDef [(sizeof(OmlMPDef) * (_MeasurementPoints + 1) )];
+
+  // define measurement points
+  unsigned int idx;
+  for (idx = 0; idx < _MeasurementPoints; ++idx)
+    createMeasurementPoint(&mp_def[idx], _OmlKeys.at(idx).first, (OmlValueT)_OmlKeys.at(idx).second);
+  createMeasurementPoint(&mp_def[idx],      "NULL",        (OmlValueT)0);
+
+  _mp_handle = omlc_add_mp (_db_filename.c_str(), mp_def); // using db_filename as tag name for measurement point
+
+  if (_mp_handle == NULL) {
+    std::cerr << "Error: could not register Measurement Point \"data\"";
+    exit (1);
+  }
+
+  result = omlc_start();
+  if (result == -1) {
+    std::cerr << "Error starting up OML measurement streams\n";
+    exit (1);
+  }
+
+  // allocate memory measurement points
+  _values = (OmlValueU*) malloc(sizeof(OmlValueU) * _MeasurementPoints);
+  memset((void*)_values, 0, sizeof(OmlValueU) * _MeasurementPoints );
+
+  // create oml key <==> (type,value) mapping
+  _KTVMap.clear();
+  for (unsigned int idx = 0; idx < _MeasurementPoints; ++idx) {
+    std::pair<OmlValueT, OmlValueU*> TV (_OmlKeys.at(idx).second, (OmlValueU*)&_values[idx] );
+    
+    std::pair<std::string, std::pair<OmlValueT, OmlValueU*> > KTV( _OmlKeys.at(idx).first, TV );
+    _KTVMap.insert( KTV );
+  }
+
+  bReady = TRUE;
+}
+
+
+void CWriteOml::set_key(std::string key_str, void* val_ptr)
+{
+
+  _KTVMapIter = _KTVMap.find(key_str);
+  if (_KTVMapIter == _KTVMap.end()) {
+    std::cerr << key_str << " not found" << std::endl;
+    return;  // key not found so return and do nothing
+  }
+
+  //key found to look at type are call appropriate oml intrinsic function
+  std::pair<OmlValueT, OmlValueU*> TV = _KTVMapIter->second;
+  switch( TV.first ) {
+  case OML_INT32_VALUE :
+    omlc_set_int32   ( *(TV.second), (int32_t) (*((int32_t*)val_ptr)));
+    break;
+
+  case OML_UINT32_VALUE :
+    omlc_set_uint32   ( *(TV.second), (uint32_t) (*((uint32_t*)val_ptr)));
+    break;
+
+  case OML_INT64_VALUE :
+    omlc_set_int64   ( *(TV.second), (int64_t) (*((int64_t*)val_ptr)));
+    break;
+  case OML_DOUBLE_VALUE :
+    omlc_set_double   ( *(TV.second), (double) (*((double*)val_ptr)));
+    break;
+  case OML_STRING_VALUE :
+    omlc_set_string( *(TV.second), (char*)val_ptr);
+    break;
+    // add other cases here
+  default :
+    std::cerr << "OML - unrecognizeg type, value: " << TV.first << " , " << TV.second << std::endl;
+    break;
+  }
+
+  return;
+}
+
+void CWriteOml::createMeasurementPoint(OmlMPDef* pOmlMPDef, std::string str, OmlValueT type)
+{
+  char* cptr;
+  if (str == "NULL") {
+    pOmlMPDef->name = NULL;
+    pOmlMPDef->param_types = type;
+  }
+  else {
+    cptr = new char[str.size()+1];
+    strcpy (cptr, str.c_str());
+    pOmlMPDef->name = cptr;
+    pOmlMPDef->param_types = type;
+  }
+}
+
+void CWriteOml::insert()
+{
+  omlc_inject (_mp_handle, _values);
+}
+
+void CWriteOml::stop()
+{
+  omlc_close();
+  free(_values);
+}

src/common/mgenApp.cpp

@@ -16 +16 @@
 #include <unistd.h>
 #include <fcntl.h>
 #endif // UNIX
+
+#include <boost/algorithm/string.hpp>
+
+
+
 MgenApp::MgenApp()
   :  mgen(GetTimerMgr(), GetSocketNotifier()),
      control_pipe(ProtoPipe::MESSAGE), control_remote(false),
@@ -56 +61 @@
             "     [queue <queueSize>][broadcast {on|off}]\n"
             "     [convert <binaryLog>][debug <debugLevel>]\n"
             "     [gpskey <gpsSharedMemoryLocation>]\n"
-            "     [boost] [reuse {on|off}]\n");
+            "     [boost] [reuse {on|off}]\n"
+        "     [oml <server:port[,omlFile]>]\n");
+
 }  // end MgenApp::Usage()
 
 
@@ -80 +87 @@
     "+gpskey",    // Override default gps shared memory location
     "+logdata",    // log optional data attribute? default ON
     "+loggpsdata", // log gps data? default ON
+    "+oml",        // oml: send data to this address:port
     NULL
 };
 
@@ -447 +455 @@
       if (!dispatcher.BoostPriority())
     fprintf(stderr,"Unable to boost process priority.\n");
     }
+    else if (!strncmp("oml", lowerCmd, len))
+    {
+      std::string temp( val );
+      std::vector<std::string> sToken;
+      boost::split(sToken, temp, boost::is_any_of(",") );
+
+#if 1
+
+      std::string omlDbFilename("oml_mgen_test");
+      std::string omlServerName;
+
+      omlServerName = sToken.at(0);
+      if (sToken.size() == 2) {
+    omlDbFilename = sToken.at(1);
+      }
+
+      mgen.oml.init(omlDbFilename, omlServerName );
+      std::vector< std::pair<std::string, OmlValueT> > _omlKeys; // this is just used here to hold all identifiers and values in one container
+                                                                 // before passing to start().
+
+      _omlKeys.push_back( std::make_pair("proto",    OML_STRING_VALUE));
+      _omlKeys.push_back( std::make_pair("flowid",   OML_INT32_VALUE));
+      _omlKeys.push_back( std::make_pair("seq_num",  OML_INT32_VALUE));
+      _omlKeys.push_back( std::make_pair("src_addr", OML_STRING_VALUE));
+      _omlKeys.push_back( std::make_pair("src_port", OML_INT32_VALUE));
+      _omlKeys.push_back( std::make_pair("dst_addr", OML_STRING_VALUE));
+      _omlKeys.push_back( std::make_pair("dst_port", OML_INT32_VALUE));
+      _omlKeys.push_back( std::make_pair("data_len", OML_INT32_VALUE));
+
+
+      //_omlKeys.push_back( std::make_pair("a_double64",    OML_DOUBLE_VALUE) );
+      //_omlKeys.push_back( std::make_pair("a____int32",   OML_INT32_VALUE));
+      //_omlKeys.push_back( std::make_pair("a___string",   OML_STRING_VALUE));
+      //_omlKeys.push_back( std::make_pair("ExecTime",    OML_DOUBLE_VALUE));
+      mgen.oml.start( _omlKeys );
+#endif
+    }
     else if (!strncmp("help", lowerCmd, len))
     {
       fprintf(stderr, "mgen: version %s\n", MGEN_VERSION);

src/common/mgenTransport.cpp

@@ -305 +305 @@
     {
     case SEND_EVENT:
       {
-          if (mgen.GetLogTx())
-          {
-              theMsg->LogSendEvent(mgen.GetLogFile(),
-                                   mgen.GetLogBinary(),
-                                   mgen.GetLocalTime(),
-                                   buffer,
-                                   mgen.GetLogFlush(),
-                                   theTime);
-          }
-          break;
+
+#if 1
+    if ( mgen.oml.isReady() == TRUE ) {
+      std::string protocol_string ( MgenEvent::GetStringFromProtocol(protocol) );
+      unsigned int u32_flowid   = theMsg->GetFlowId();
+      unsigned int u32_seqnum   = theMsg->GetSeqNum();
+      std::string str_srcaddr   = theMsg->GetSrcAddr().GetHostString();
+      unsigned int u32_srcport  = theMsg->GetSrcAddr().GetPort();
+      std::string str_dstaddr   = theMsg->GetDstAddr().GetHostString();
+      unsigned int u32_dstport  = theMsg->GetDstAddr().GetPort();
+      unsigned int u32_datalen  = theMsg->GetMgenMsgLen();
+
+      mgen.oml.set_key("proto",     (void*)protocol_string.c_str() );
+      mgen.oml.set_key("flowid",    (void*)&u32_flowid );
+      mgen.oml.set_key("seq_num",   (void*)&u32_seqnum );
+      mgen.oml.set_key("src_addr",  (void*)str_srcaddr.c_str() );
+      mgen.oml.set_key("src_port",  (void*)&u32_srcport );
+      mgen.oml.set_key("dst_addr",  (void*)str_dstaddr.c_str() );
+      mgen.oml.set_key("dst_port",  (void*)&u32_dstport );
+      mgen.oml.set_key("data_len",  (void*)&u32_datalen );
+      mgen.oml.insert();
+    }
+#endif
+
+    if (mgen.GetLogTx())
+        {
+      theMsg->LogSendEvent(mgen.GetLogFile(),
+                   mgen.GetLogBinary(),
+                   mgen.GetLocalTime(),
+                   buffer,
+                   mgen.GetLogFlush(),
+                   theTime);
+        }
+    break;
       }
     case RECV_EVENT:
       {
@@ -328 +352 @@
                                mgen.GetLogFlush(),
                                theTime);
 
+#if 1
+      if ( mgen.oml.isReady() == TRUE ) {
+        std::string protocol_string ( MgenEvent::GetStringFromProtocol(protocol) );
+        unsigned int u32_flowid   = theMsg->GetFlowId();
+        unsigned int u32_seqnum   = theMsg->GetSeqNum();
+        std::string str_srcaddr   = theMsg->GetSrcAddr().GetHostString();
+        unsigned int u32_srcport  = theMsg->GetSrcAddr().GetPort();
+        std::string str_dstaddr   = theMsg->GetDstAddr().GetHostString();
+        unsigned int u32_dstport  = theMsg->GetDstAddr().GetPort();
+        unsigned int u32_datalen  = theMsg->GetMsgLen();
+
+        mgen.oml.set_key("proto",     (void*)protocol_string.c_str() );
+        mgen.oml.set_key("flowid",    (void*)&u32_flowid );
+        mgen.oml.set_key("seq_num",   (void*)&u32_seqnum );
+        mgen.oml.set_key("src_addr",  (void*)str_srcaddr.c_str() );
+        mgen.oml.set_key("src_port",  (void*)&u32_srcport );
+        mgen.oml.set_key("dst_addr",  (void*)str_dstaddr.c_str() );
+        mgen.oml.set_key("dst_port",  (void*)&u32_dstport );
+        mgen.oml.set_key("data_len",  (void*)&u32_datalen );
+        mgen.oml.insert();
+      }
+#endif
+
+
           // Don't we want rapr to get the message regardless of logging??
           // Could this possibly have been broken too? strange... ljt
           if (mgen.GetController())