Line data    Source code

       1             : /*
       2             :  * Copyright (c) 2013 Juniper Networks, Inc. All rights reserved.
       3             :  * Copyright Contributors to the OpenSDN project
       4             :  */
       5             : 
       6             : #include <utility>
       7             : #include <vector>
       8             : #include <map>
       9             : #include <boost/bind/bind.hpp>
      10             : #include "base/timer.h"
      11             : #include <boost/date_time/posix_time/posix_time.hpp>
      12             : #include <boost/assign/list_of.hpp>
      13             : 
      14             : #include <sandesh/sandesh_trace.h>
      15             : #include <sandesh/sandesh_session.h>
      16             : #include <sandesh/sandesh_ctrl_types.h>
      17             : #include <sandesh/common/vns_types.h>
      18             : #include <sandesh/common/vns_constants.h>
      19             : #include <sandesh/sandesh_uve_types.h>
      20             : #include <sandesh/sandesh_message_builder.h>
      21             : 
      22             : #include "OpServerProxy.h"
      23             : #include "db_handler.h"
      24             : #include "collector.h"
      25             : #include "generator.h"
      26             : #include "viz_collector.h"
      27             : #include "viz_sandesh.h"
      28             : #include <analytics/viz_types.h>
      29             : #include "vizd_table_desc.h"
      30             : 
      31             : extern SandeshTraceBufferPtr UVETraceBuf;
      32             : 
      33             : using std::string;
      34             : using std::pair;
      35             : using std::vector;
      36             : using std::map;
      37             : using namespace boost::placeholders;
      38             : 
      39             : #define GENERATOR_LOG(_Level, _Msg)                                            \
      40             :     do {                                                                       \
      41             :         if (LoggingDisabled()) break;                                          \
      42             :         log4cplus::Logger _Xlogger = log4cplus::Logger::getRoot();             \
      43             :         if (_Xlogger.isEnabledFor(log4cplus::_Level##_LOG_LEVEL)) {            \
      44             :             log4cplus::tostringstream _Xbuf;                                   \
      45             :             _Xbuf << ToString() << ": " << __func__ << ": " << _Msg;           \
      46             :             _Xlogger.forcedLog(log4cplus::_Level##_LOG_LEVEL,                  \
      47             :                                _Xbuf.str());                                   \
      48             :         }                                                                      \
      49             :     } while (false)
      50             : 
      51           0 : void Generator::UpdateStatistics(const VizMsg *vmsg) {
      52           0 :     std::scoped_lock lock(smutex_);
      53           0 :     statistics_.Update(vmsg);
      54           0 : }
      55             : 
      56           0 : void Generator::GetStatistics(vector<SandeshStats> *ssv) const {
      57           0 :     std::scoped_lock lock(smutex_);
      58           0 :     statistics_.Get(ssv);
      59           0 : }
      60             : 
      61           0 : void Generator::GetStatistics(vector<SandeshLogLevelStats> *lsv) const {
      62           0 :     std::scoped_lock lock(smutex_);
      63           0 :     statistics_.Get(lsv);
      64           0 : }
      65             : 
      66           0 : void Generator::SendSandeshMessageStatistics() {
      67           0 :     vector<SandeshMessageInfo> smv;
      68             :     {
      69           0 :         std::scoped_lock lock(smutex_);
      70           0 :         statistics_.Get(&smv);
      71           0 :     }
      72           0 :     SandeshMessageStat * snh = SANDESH_MESSAGE_STAT_CREATE();
      73           0 :     snh->set_name(ToString());
      74           0 :     snh->set_msg_info(smv);
      75           0 :     SANDESH_MESSAGE_STAT_SEND_SANDESH(snh);
      76           0 : }
      77             :     
      78           0 : bool Generator::ReceiveSandeshMsg(const VizMsg *vmsg, bool rsc) {
      79           0 :     UpdateStatistics(vmsg);
      80           0 :     return ProcessRules(vmsg, rsc);
      81             : }
      82             : 
      83             : // SandeshGenerator
      84           0 : SandeshGenerator::SandeshGenerator(Collector * const collector, VizSession *session,
      85             :         SandeshStateMachine *state_machine, const string &source,
      86             :         const string &module, const string &instance_id,
      87             :         const string &node_type,
      88           0 :         DbHandlerPtr global_db_handler) :
      89             :         Generator(),
      90           0 :         collector_(collector),
      91           0 :         state_machine_(state_machine),
      92           0 :         viz_session_(session),
      93           0 :         instance_id_(instance_id),
      94           0 :         node_type_(node_type),
      95           0 :         source_(source),
      96           0 :         module_(module),
      97           0 :         name_(source + ":" + node_type_ + ":" + module + ":" + instance_id_),
      98           0 :         instance_(session->GetSessionInstance()),
      99           0 :         process_rules_cb_(
     100             :             boost::bind(&SandeshGenerator::ProcessRulesCb, this, _1)),
     101           0 :         sm_defer_timer_(NULL),
     102           0 :         sm_defer_timer_expiry_time_usec_(0),
     103           0 :         sm_defer_time_msec_(0) {
     104             :         //Use collector db_handler
     105           0 :         db_handler_ = global_db_handler;
     106           0 :         disconnected_ = false;
     107           0 :         gen_attr_.set_connects(1);
     108           0 :         gen_attr_.set_connect_time(UTCTimestampUsec());
     109             :         // Update state machine
     110           0 :         state_machine_->SetGeneratorKey(name_);
     111           0 :         CreateStateMachineDeferTimer();
     112           0 : }
     113             : 
     114           0 : SandeshGenerator::~SandeshGenerator() {
     115           0 :     DeleteStateMachineDeferTimer();
     116           0 : }
     117             : 
     118           0 : void SandeshGenerator::set_session(VizSession *session) {
     119           0 :     viz_session_ = session;
     120           0 :     instance_ = session->GetSessionInstance();
     121           0 :     session->set_generator(this);
     122           0 : }
     123             : 
     124           0 : void SandeshGenerator::TimerErrorHandler(string name, string error) {
     125           0 :     GENERATOR_LOG(ERROR, name + " error: " + error);
     126           0 : }
     127             : 
     128           0 : void SandeshGenerator::ReceiveSandeshCtrlMsg(uint32_t connects) {
     129             :     // This is a control message during SandeshGenerator-Collector negotiation
     130           0 :     ModuleServerState ginfo;
     131           0 :     GetGeneratorInfo(ginfo);
     132           0 :     SandeshModuleServerTrace::Send(ginfo);
     133             :     // Setup state machine watermarks
     134           0 :     std::vector<Sandesh::QueueWaterMarkInfo> wm_info;
     135           0 :     collector_->GetSmQueueWaterMarkInfo(wm_info);
     136           0 :     for (size_t i = 0; i < wm_info.size(); i++) {
     137           0 :         state_machine_->SetQueueWaterMarkInfo(wm_info[i]);
     138             :     }
     139           0 : }
     140             : 
     141           0 : void SandeshGenerator::DisconnectSession(VizSession *vsession) {
     142           0 :     std::scoped_lock lock(mutex_);
     143           0 :     GENERATOR_LOG(INFO, "Session:" << vsession->ToString());
     144           0 :     if (vsession == viz_session_) {
     145           0 :         disconnected_ = true;
     146             :         // This SandeshGenerator's session is now gone.
     147             :         // Delete all its UVEs
     148           0 :         uint32_t tmp = gen_attr_.get_resets();
     149           0 :         gen_attr_.set_resets(tmp+1);
     150           0 :         gen_attr_.set_reset_time(UTCTimestampUsec());
     151           0 :         state_machine_->ResetQueueWaterMarkInfo();
     152           0 :         StopStateMachineDeferTimer();
     153           0 :         DeleteStateMachineDeferTimer();
     154           0 :         sm_defer_timer_expiry_time_usec_ = 0;
     155           0 :         sm_defer_time_msec_ = 0;
     156           0 :         viz_session_ = NULL;
     157           0 :         state_machine_ = NULL;
     158           0 :         vsession->set_generator(NULL);
     159           0 :         collector_->GetOSP()->DeleteUVEs(source_, module_, 
     160           0 :                                          node_type_, instance_id_);
     161           0 :         ModuleServerState ginfo;
     162           0 :         GetGeneratorInfo(ginfo);
     163           0 :         SandeshModuleServerTrace::Send(ginfo);
     164           0 :     } else {
     165           0 :         GENERATOR_LOG(ERROR, "Disconnect for session:" << vsession->ToString() <<
     166             :                 ", generator session:" << viz_session_->ToString());
     167             :     }
     168           0 : }
     169             : 
     170           0 : bool SandeshGenerator::StateMachineDeferTimerExpired() {
     171           0 :     std::scoped_lock lock(mutex_);
     172           0 :     sm_defer_timer_expiry_time_usec_ = UTCTimestampUsec();
     173           0 :     if (state_machine_) {
     174           0 :         state_machine_->SetDeferDequeue(false);
     175             :     }
     176           0 :     return false;
     177           0 : }
     178             : 
     179           0 : void SandeshGenerator::CreateStateMachineDeferTimer() {
     180             :     // Run in the context of sandesh state machine task
     181           0 :     assert(sm_defer_timer_ == NULL);
     182           0 :     sm_defer_timer_ = TimerManager::CreateTimer(
     183           0 :         *collector_->event_manager()->io_service(),
     184           0 :         "SandeshGenerator SM Defer Timer: " + name_,
     185           0 :         state_machine_->connection()->GetTaskId(), instance_);
     186           0 : }
     187             : 
     188           0 : void SandeshGenerator::StartStateMachineDeferTimer(int time_msec) {
     189           0 :     sm_defer_timer_->Start(time_msec,
     190             :             boost::bind(
     191             :                 &SandeshGenerator::StateMachineDeferTimerExpired, this),
     192             :             boost::bind(&SandeshGenerator::TimerErrorHandler, this, _1, _2));
     193           0 : }
     194             : 
     195           0 : void SandeshGenerator::StopStateMachineDeferTimer() {
     196           0 :     assert(sm_defer_timer_->Cancel());
     197           0 : }
     198             : 
     199           0 : void SandeshGenerator::DeleteStateMachineDeferTimer() {
     200           0 :     TimerManager::DeleteTimer(sm_defer_timer_);
     201           0 :     sm_defer_timer_ = NULL;
     202           0 : }
     203             : 
     204           0 : bool SandeshGenerator::IsStateMachineDeferTimerRunningUnlocked() const {
     205           0 :     if (sm_defer_timer_) {
     206           0 :         return sm_defer_timer_->running();
     207             :     }
     208           0 :     return false;
     209             : }
     210             : 
     211           0 : bool SandeshGenerator::IsStateMachineDeferTimerRunning() const {
     212           0 :     std::scoped_lock lock(mutex_);
     213           0 :     return IsStateMachineDeferTimerRunningUnlocked();
     214           0 : }
     215             : 
     216           0 : int SandeshGenerator::GetStateMachineDeferTimeMSec() const {
     217           0 :     std::scoped_lock lock(mutex_);
     218           0 :     return sm_defer_time_msec_;
     219           0 : }
     220             : 
     221           7 : int GetDeferTimeMSec(uint64_t event_time_usec,
     222             :     uint64_t last_expiry_time_usec, uint64_t last_defer_time_usec) {
     223             :     // If this is the first time, then defer the state machine with
     224             :     // initial defer time
     225           7 :     if (last_defer_time_usec == 0 || last_expiry_time_usec == 0) {
     226           2 :         return SandeshGenerator::kInitialSmDeferTimeMSec;
     227             :     }
     228           5 :     assert(event_time_usec >= last_expiry_time_usec);
     229           5 :     uint64_t time_since_expiry_usec(event_time_usec - last_expiry_time_usec);
     230             :     // We will double the defer time if we get a back pressure
     231             :     // event within 2 * last defer time. If the back pressure
     232             :     // event is between 2 * last defer time and 4 * last defer
     233             :     // time, then the defer time will be same as the current
     234             :     // defer time. If the back pressure event is after  4 * last
     235             :     // defer time, then we will reset the defer time to the
     236             :     // initial defer time
     237           5 :     if (time_since_expiry_usec <= 2 * last_defer_time_usec) {
     238           3 :         uint64_t ndefer_time_msec((2 * last_defer_time_usec)/1000);
     239           6 :         return std::min(ndefer_time_msec,
     240           3 :             static_cast<uint64_t>(SandeshGenerator::kMaxSmDeferTimeMSec));
     241           2 :     } else if ((2 * last_defer_time_usec <= time_since_expiry_usec) &&
     242           2 :         (time_since_expiry_usec <= 4 * last_defer_time_usec)) {
     243           1 :         return last_defer_time_usec/1000;
     244             :     } else {
     245           1 :         return SandeshGenerator::kInitialSmDeferTimeMSec;
     246             :     }
     247             : }
     248             : 
     249           0 : void SandeshGenerator::ProcessRulesCb(GenDb::DbOpResult::type dresult) {
     250           0 :     std::scoped_lock lock(mutex_);
     251           0 :     if (dresult == GenDb::DbOpResult::BACK_PRESSURE) {
     252           0 :         if (state_machine_) {
     253             :             // If state mchine defer timer is running just return to
     254             :             // avoid increasing the defer time more than once every
     255             :             // timer expiry
     256           0 :             if (IsStateMachineDeferTimerRunningUnlocked()) {
     257           0 :                 return;
     258             :             }
     259           0 :             state_machine_->SetDeferDequeue(true);
     260           0 :             uint64_t now_usec(UTCTimestampUsec());
     261           0 :             int defer_time_msec(GetDeferTimeMSec(now_usec,
     262           0 :                 sm_defer_timer_expiry_time_usec_, sm_defer_time_msec_ * 1000));
     263           0 :             sm_defer_time_msec_ = defer_time_msec;
     264           0 :             StartStateMachineDeferTimer(sm_defer_time_msec_);
     265             :         }
     266             :     }
     267           0 : }
     268             : 
     269           0 : bool SandeshGenerator::ProcessRules(const VizMsg *vmsg, bool rsc) {
     270           0 :     return collector_->ProcessSandeshMsgCb()(vmsg, rsc, GetDbHandler(),
     271           0 :         process_rules_cb_);
     272             : }
     273             : 
     274           0 : bool SandeshGenerator::GetSandeshStateMachineQueueCount(
     275             :     uint64_t &queue_count) const {
     276           0 :     if (!state_machine_) {
     277             :         // Return 0 so that last stale value is not displayed
     278           0 :         queue_count = 0;
     279           0 :         return true;
     280             :     }
     281           0 :     return state_machine_->GetQueueCount(queue_count);
     282             : }
     283             : 
     284           0 : bool SandeshGenerator::GetSandeshStateMachineDropLevel(
     285             :     std::string &drop_level) const {
     286           0 :     if (!state_machine_) {
     287           0 :         return false;
     288             :     }
     289           0 :     return state_machine_->GetMessageDropLevel(drop_level);
     290             : }
     291             : 
     292           0 : bool SandeshGenerator::GetSandeshStateMachineStats(
     293             :                     SandeshStateMachineStats &sm_stats,
     294             :                     SandeshGeneratorBasicStats &sm_msg_stats) const {
     295           0 :     if (!state_machine_) {
     296           0 :         return false;
     297             :     }
     298           0 :     return state_machine_->GetStatistics(sm_stats, sm_msg_stats);
     299             : }
     300             : 
     301           0 : void SandeshGenerator::GetGeneratorInfo(ModuleServerState &genlist) const {
     302           0 :     vector<GeneratorInfo> giv;
     303           0 :     GeneratorInfo gi;
     304           0 :     gi.set_hostname(Sandesh::source());
     305           0 :     gi.set_gen_attr(gen_attr_);
     306           0 :     giv.push_back(gi);
     307           0 :     genlist.set_generator_info(giv);
     308           0 :     genlist.set_name(source() + ":" + node_type_ + ":" + module() + ":" +
     309           0 :         instance_id_);
     310           0 : }
     311             : 
     312           0 : const std::string SandeshGenerator::State() const {
     313           0 :     if (state_machine_) {
     314           0 :         return state_machine_->StateName();
     315             :     }
     316           0 :     return "Disconnected";
     317             : }
     318             : 
     319           0 : void SandeshGenerator::ConnectSession(VizSession *session,
     320             :     SandeshStateMachine *state_machine) {
     321           0 :     std::scoped_lock lock(mutex_);
     322           0 :     set_session(session);
     323           0 :     set_state_machine(state_machine);
     324           0 :     disconnected_ = false;
     325           0 :     uint32_t tmp = gen_attr_.get_connects();
     326           0 :     gen_attr_.set_connects(tmp+1);
     327           0 :     gen_attr_.set_connect_time(UTCTimestampUsec());
     328           0 :     CreateStateMachineDeferTimer();
     329           0 : }
     330             : 
     331           0 : void SandeshGenerator::SetDbQueueWaterMarkInfo(
     332             :     Sandesh::QueueWaterMarkInfo &wm) {
     333           0 :     if (!GetDbHandler()) {
     334           0 :         return;
     335             :     }
     336           0 :     bool high(boost::get<2>(wm));
     337           0 :     bool defer_undefer(boost::get<3>(wm));
     338           0 :     boost::function<void (void)> cb;
     339           0 :     if (high && defer_undefer) {
     340           0 :         cb = boost::bind(&SandeshStateMachine::SetDeferDequeue,
     341           0 :                 state_machine_, true);
     342           0 :     } else if (!high && defer_undefer) {
     343             :         cb = boost::bind(&SandeshStateMachine::SetDeferDequeue,
     344           0 :                 state_machine_, false);
     345             :     }
     346           0 :     GetDbHandler()->SetDbQueueWaterMarkInfo(wm, cb);
     347           0 : }
     348             : 
     349           0 : void SandeshGenerator::ResetDbQueueWaterMarkInfo() {
     350           0 :     if (!GetDbHandler()) {
     351           0 :         return;
     352             :     }
     353           0 :     GetDbHandler()->ResetDbQueueWaterMarkInfo();
     354             : }
     355             : 
     356           0 : void SandeshGenerator::SetSmQueueWaterMarkInfo(
     357             :     Sandesh::QueueWaterMarkInfo &wm) {
     358           0 :     if (state_machine_) {
     359           0 :         state_machine_->SetQueueWaterMarkInfo(wm);
     360             :     }
     361           0 : }
     362             : 
     363           0 : void SandeshGenerator::ResetSmQueueWaterMarkInfo() {
     364           0 :     if (state_machine_) {
     365           0 :         state_machine_->ResetQueueWaterMarkInfo();
     366             :     }
     367           0 : }
     368             : 
     369             : // SyslogGenerator
     370           0 : SyslogGenerator::SyslogGenerator(SyslogListeners *const listeners,
     371           0 :         const string &source, const string &module) :
     372             :           Generator(),
     373           0 :           syslog_(listeners),
     374           0 :           source_(source),
     375           0 :           module_(module),
     376           0 :           name_(source + ":" + module),
     377           0 :           db_handler_(listeners->GetDbHandler()) {
     378           0 : }
     379             : 
     380           0 : bool SyslogGenerator::ProcessRules(const VizMsg *vmsg, bool rsc) {
     381           0 :     return syslog_->ProcessSandeshMsgCb()(vmsg, rsc, GetDbHandler(),
     382           0 :         GenDb::GenDbIf::DbAddColumnCb());
     383             : }