sicspsi/mongolog.c

/*
  This is a module which implements output of the log to a mongodb NoSQL database. 
  The database is much easier for searching the log

  copyright: see file COPYRIGHT

  Mark Koennecke, February 2016
*/

#include <stdio.h>
#include <logv2.h>
#include <bson.h>
#include <mongoc.h>
#include <sicsutil.h>
#include <sics.h>
#include "sicslogquery.h"

static char mongoURL[1024];
static char inst[132];

static mongoc_client_t *client;
static mongoc_collection_t *collection;

static void MongoCallback(unsigned int severity, const char *timeStamp, 
			    const char *subsystem, 
		   const char *message, void *userData);

/*-----------------------------------------------------------------------------------*/
static void MongoClose(void *data)
{
  RemoveLogCallback(MongoCallback);
  if(collection != NULL){
    mongoc_collection_destroy (collection);
    mongoc_client_destroy (client);
    mongoc_cleanup ();
  }
  client = NULL;
  collection = NULL;
}
/*----------------------------------------------------------------------------------
  The callback function for entering data into the log
  ------------------------------------------------------------------------------------*/
static void MongoCallback(unsigned int severity, const char *timeStamp, 
			    const char *subsystem, 
			    const char *message, void *userData)
{
  double tVal = DoubleTime();
  int status;
  bson_t *doc;
  bson_error_t error;
  bson_oid_t oid;

  if(client == NULL){
    return;
  }

  if(logFilter(severity,subsystem,message) == 1) {
    return;
  }

  doc = bson_new ();
  bson_oid_init (&oid, NULL);
  BSON_APPEND_OID (doc, "_id", &oid);
  BSON_APPEND_DOUBLE(doc,"timestamp",tVal);
  BSON_APPEND_INT32(doc,"severity",severity);
  BSON_APPEND_UTF8(doc,"timetext",timeStamp); 
  BSON_APPEND_UTF8(doc,"sub",subsystem); 
  BSON_APPEND_UTF8(doc,"message",message); 

   if (!mongoc_collection_insert (collection, MONGOC_INSERT_NONE, doc, NULL, &error)) {
        fprintf (stdout, "%s\n", error.message);
	MongoClose(NULL);
	Log(FATAL,"sys", "Lost connection to mongo server at %s", mongoURL);
   }

   bson_destroy (doc);
}
/*------------------------------------------------------------------------------------*/
static int MongoConfigAction(SConnection * pCon, SicsInterp * pSics,
                            void *pData, int argc, char *argv[])
{
  int status;

  if(argc < 2) {
    SCWrite(pCon,"ERROR: need keyword", eError);
    return 0;  
  }

   strtolower(argv[1]);
   if(strcmp(argv[1],"open") == 0){
     if(argc < 3){
       SCWrite(pCon,"ERROR: need mongourl and instrument  parameters for open",eError);
       return 0;
     }
     strncpy(mongoURL,argv[2],sizeof(mongoURL));
     strncpy(inst, argv[3],sizeof(inst));
     mongoc_init();
     client = mongoc_client_new (mongoURL);
     if(!client){
       SCWrite(pCon,"ERROR: failed to connect to mongodb, bad URL?", eError);
       return 0;
     }
     collection = mongoc_client_get_collection (client, inst, "log");
     if(!collection){
       SCWrite(pCon,"ERROR: failed to connect link to mongo DB", eError);
       return 0;
     }
     RegisterLogCallback(MongoCallback,NULL);
     SCSendOK(pCon);
   } else if(strcmp(argv[1],"reopen") == 0) {
     mongoc_init();
     client = mongoc_client_new (mongoURL);
     if(!client){
       SCWrite(pCon,"ERROR: failed to connect to mongodb, bad URL?", eError);
       return 0;
     }
     collection = mongoc_client_get_collection (client, inst, "log");
     if(!collection){
       SCWrite(pCon,"ERROR: failed to connect link to mongo DB", eError);
       return 0;
     }
     RegisterLogCallback(MongoCallback,NULL);
     SCSendOK(pCon);
   }else if(strcmp(argv[1],"status") == 0){
     if(client == NULL){
       SCWrite(pCon,"Mongo is disconnected", eValue);
     } else {
       SCWrite(pCon,"Mongo is connected", eValue);
     }
   } else if(strcmp(argv[1],"close") == 0) {
       MongoClose(NULL);
       SCSendOK(pCon);
   } else if(strcmp(argv[1],"filename") == 0) {
     SCPrintf(pCon,eValue,"mongodb = %s", mongoURL);
   } else {
     SCPrintf(pCon,eError,"ERROR: keyword %s not recognised", argv[1]);
     return 0;
   }
   return 1;
}
/*----------------------------------------------------------------------------*/
static void QueryCallback(const bson_t *doc, void *userData)
{
  pDynString result = (pDynString)userData;
  bson_iter_t iter;
  unsigned int severity;
  uint32_t length;
  const char *message, *timeText, *sub;
  char sevBuf[20];

  bson_iter_init(&iter,doc);
  bson_iter_find(&iter,"timetext");
  timeText = bson_iter_utf8(&iter,&length);
  DynStringConcat(result,(char *)timeText);
  DynStringConcatChar(result,' ');
  bson_iter_init(&iter,doc);
  bson_iter_find(&iter,"severity");
  severity = bson_iter_int32(&iter);
  formatSeverity(severity,sevBuf,sizeof(sevBuf));
  DynStringConcat(result,(char *)sevBuf);
  DynStringConcatChar(result,' ');
  bson_iter_find(&iter,"sub");
  sub = bson_iter_utf8(&iter,&length);
  DynStringConcat(result,(char *)sub);
  DynStringConcatChar(result,' ');
  bson_iter_find(&iter,"message");
  message = bson_iter_utf8(&iter,&length);
  DynStringConcat(result,(char *)message);
  DynStringConcatChar(result,'\n');
}
/*-----------------------------------------------------------------------------*/
static int MongoQueryAction(SConnection * pCon, SicsInterp * pSics,
                            void *pData, int argc, char *argv[])
{
  pDynString result; 
  char *error = NULL;
  int status;

  sicslogSetup(mongoURL,inst);
  result = CreateDynString(132,132);

  status = sicslogQuery(argc,argv,QueryCallback,result);
  if(status != 0){
    error = sicslogGetError();
    if(strstr(error,"Options") != NULL){
      SCPrintf(pCon,eValue,"%s", error);
    } else {
      SCPrintf(pCon,eError,"Error %s querying mongodb", error);
    }
  } else {
    SCPureSockWrite(pCon,GetCharArray(result),eLog);
  }
  DeleteDynString(result);
  return 1;
}
/*-----------------------------------------------------------------------------*/
void MongoLogInit(void)
{
  AddCommand(pServ->pSics,"mongoconfig", MongoConfigAction,
               MongoClose, NULL);
  AddCommand(pServ->pSics,"showlog", MongoQueryAction,
               MongoClose, NULL);
}