diff --git a/apps/monitoring/Monitoring.cpp b/apps/monitoring/Monitoring.cpp index 452cea14..88a5ceab 100644 --- a/apps/monitoring/Monitoring.cpp +++ b/apps/monitoring/Monitoring.cpp @@ -27,10 +27,14 @@ #include "Monitoring.h" +#include "AmConfigReader.h" +#include "AmEventDispatcher.h" + #include "log.h" #include #include +#include //EXPORT_PLUGIN_CLASS_FACTORY(Monitor, MOD_NAME); extern "C" void* plugin_class_create() @@ -42,6 +46,7 @@ extern "C" void* plugin_class_create() } Monitor* Monitor::_instance=0; +unsigned int Monitor::gcInterval = 10; Monitor* Monitor::instance() { @@ -51,7 +56,7 @@ Monitor* Monitor::instance() } Monitor::Monitor(const string& name) - : AmDynInvokeFactory(MOD_NAME) { + : AmDynInvokeFactory(MOD_NAME), gc_thread(NULL) { } Monitor::~Monitor() { @@ -59,6 +64,23 @@ Monitor::~Monitor() { int Monitor::onLoad() { // todo: if GC configured, start thread + AmConfigReader cfg; + if(cfg.loadFile(AmConfig::ModConfigPath + string(MOD_NAME ".conf"))) { + DBG("monitoring not starting garbage collector\n"); + return 0; + } + + if (cfg.getParameter("run_garbage_collector","no") == "yes") { + gcInterval = cfg.getParameterInt("garbage_collector_interval", 10); + DBG("Running garbage collection for monitoring every %u seconds\n", + gcInterval); + gc_thread.reset(new MonitorGarbageCollector()); + gc_thread->start(); + AmEventDispatcher::instance()->addEventQueue("monitoring_gc", gc_thread.get()); +// // add garbage collector to garbage collector... +// AmThreadWatcher::instance()->add(gc_thread); + } + return 0; } @@ -189,12 +211,21 @@ void Monitor::clear(const AmArg& args, AmArg& ret) { } void Monitor::clearFinished(const AmArg& args, AmArg& ret) { - for (int i=0;i::iterator it= logs[i].log.begin(); while (it != logs[i].log.end()) { - if (it->second.finished > 0) { + if (it->second.finished && + it->second.finished <= now) { std::map::iterator d_it = it; it++; logs[i].log.erase(d_it); @@ -204,8 +235,6 @@ void Monitor::clearFinished(const AmArg& args, AmArg& ret) { } logs[i].log_lock.unlock(); } - ret.push(0); - ret.push("OK"); } void Monitor::get(const AmArg& args, AmArg& ret) { @@ -218,10 +247,28 @@ void Monitor::get(const AmArg& args, AmArg& ret) { bucket.log_lock.unlock(); } +void Monitor::getAttribute(const AmArg& args, AmArg& ret) { + assertArgCStr(args[0]); + string attr_name = args[0].asCStr(); + for (int i=0;i::iterator it= + logs[i].log.begin();it != logs[i].log.end();it++) { + ret.push(AmArg()); + AmArg& val = ret.get(ret.size()-1); + val.push(AmArg(it->first.c_str())); + val.push(it->second.info[attr_name]); + } + logs[i].log_lock.unlock(); + } +} + + #define DEF_GET_ATTRIB_FUNC(func_name, cond) \ void Monitor::func_name(const AmArg& args, AmArg& ret) { \ assertArgCStr(args[0]); \ string attr_name = args[0].asCStr(); \ + time_t now = time(0); \ for (int i=0;i::iterator it= \ @@ -237,10 +284,10 @@ void Monitor::get(const AmArg& args, AmArg& ret) { } \ } -DEF_GET_ATTRIB_FUNC(getAttribute, true) -DEF_GET_ATTRIB_FUNC(getAttributeActive, (!it->second.finished)) -DEF_GET_ATTRIB_FUNC(getAttributeFinished, (it->second.finished)) - +DEF_GET_ATTRIB_FUNC(getAttributeActive, (!(it->second.finished && + it->second.finished <= now))) +DEF_GET_ATTRIB_FUNC(getAttributeFinished,(it->second.finished && + it->second.finished <= now)) #undef DEF_GET_ATTRIB_FUNC void Monitor::listAll(const AmArg& args, AmArg& ret) { @@ -315,11 +362,14 @@ void Monitor::listByRegex(const AmArg& args, AmArg& ret) { } void Monitor::listFinished(const AmArg& args, AmArg& ret) { - for (int i=0;i::iterator it= logs[i].log.begin(); it != logs[i].log.end(); it++) { - if (it->second.finished > 0) + if (it->second.finished && + it->second.finished <= now) ret.push(AmArg(it->first.c_str())); } logs[i].log_lock.unlock(); @@ -328,11 +378,14 @@ void Monitor::listFinished(const AmArg& args, AmArg& ret) { void Monitor::listActive(const AmArg& args, AmArg& ret) { - for (int i=0;i::iterator it= logs[i].log.begin(); it != logs[i].log.end(); it++) { - if (!it->second.finished) + if (!(it->second.finished && + it->second.finished <= now)) ret.push(AmArg(it->first.c_str())); } logs[i].log_lock.unlock(); @@ -348,3 +401,29 @@ LogBucket& Monitor::getLogBucket(const string& call_id) { return logs[c % NUM_LOG_BUCKETS]; } + +void MonitorGarbageCollector::run() { + DBG("running MonitorGarbageCollector thread\n"); + running.set(true); + while (running.get()) { + sleep(Monitor::gcInterval); + Monitor::instance()->clearFinished(); + } + DBG("MonitorGarbageCollector thread ends\n"); + AmEventDispatcher::instance()->delEventQueue("monitoring_gc"); +} + +void MonitorGarbageCollector::postEvent(AmEvent* e) { + AmSystemEvent* sys_ev = dynamic_cast(e); + if (sys_ev && + sys_ev->sys_event == AmSystemEvent::ServerShutdown) { + DBG("stopping MonitorGarbageCollector thread\n"); + running.set(false); + return; + } + + WARN("received unknown event\n"); +} + +void MonitorGarbageCollector::on_stop() { +} diff --git a/apps/monitoring/Monitoring.h b/apps/monitoring/Monitoring.h index 55e74227..c5ae66a0 100644 --- a/apps/monitoring/Monitoring.h +++ b/apps/monitoring/Monitoring.h @@ -29,6 +29,7 @@ #define _MONITORING_H_ #include +#include #include "AmThread.h" #include "AmApi.h" @@ -49,18 +50,21 @@ struct LogBucket { AmMutex log_lock; std::map log; }; +class MonitorGarbageCollector; class Monitor : public AmDynInvokeFactory, - public AmDynInvoke - + public AmDynInvoke { static Monitor* _instance; + std::auto_ptr gc_thread; LogBucket logs[NUM_LOG_BUCKETS]; LogBucket& getLogBucket(const string& call_id); + void clearFinished(); + void log(const AmArg& args, AmArg& ret); void logAdd(const AmArg& args, AmArg& ret); void markFinished(const AmArg& args, AmArg& ret); @@ -88,15 +92,20 @@ class Monitor void invoke(const string& method, const AmArg& args, AmArg& ret); int onLoad(); + static unsigned int gcInterval; + + friend class MonitorGarbageCollector; }; -/* -class MonitorGarbageCollector : public AmThread { +class MonitorGarbageCollector +: public AmThread, + public AmEventQueueInterface + { + AmSharedVar running; + public: void run(); void on_stop(); - + void postEvent(AmEvent* e); }; -*/ - #endif diff --git a/apps/monitoring/etc/monitoring.conf b/apps/monitoring/etc/monitoring.conf new file mode 100644 index 00000000..02677859 --- /dev/null +++ b/apps/monitoring/etc/monitoring.conf @@ -0,0 +1,14 @@ + +#run_garbage_collector=[yes | no] +# +# run garbage collection on expired session info? +# Default: no +# +#run_garbage_collector = yes + +#garbage_collector_interval=10 +# +# run garbage collection every n seconds +# Default: 10 +# +#garbage_collector_interval = 20 \ No newline at end of file diff --git a/doc/Readme.monitoring b/doc/Readme.monitoring index 19696efe..d1abda91 100644 --- a/doc/Readme.monitoring +++ b/doc/Readme.monitoring @@ -25,6 +25,8 @@ monitoring must be compile time enabled in Makefile.defs by setting USE_MONITORING = yes and the monitoring module needs to be loaded. +In monitoring.conf the option can be set to run a garbage collector thread. +This will remove all info about finished sessions preiodically. DI API ------