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 <boost/assign/list_of.hpp>
       7             : #include <boost/lexical_cast.hpp>
       8             : #include <boost/bind/bind.hpp>
       9             : #include <boost/assign.hpp>
      10             : #include <boost/assign/list_of.hpp>
      11             : #include <boost/array.hpp>
      12             : #include <boost/uuid/name_generator.hpp>
      13             : 
      14             : #include "base/logging.h"
      15             : #include "base/task.h"
      16             : #include "base/parse_object.h"
      17             : #include "base/address_util.h"
      18             : #include <base/connection_info.h>
      19             : #include "io/event_manager.h"
      20             : 
      21             : #include <sandesh/sandesh_ctrl_types.h>
      22             : #include <sandesh/sandesh_uve_types.h>
      23             : #include <sandesh/sandesh_statistics.h>
      24             : #include <sandesh/sandesh_session.h>
      25             : #include <sandesh/sandesh_connection.h>
      26             : #include <sandesh/sandesh_state_machine.h>
      27             : #include <sandesh/sandesh_message_builder.h>
      28             : #include "collector.h"
      29             : #include "viz_collector.h"
      30             : #include "viz_sandesh.h"
      31             : #include <analytics_types.h>
      32             : 
      33             : using std::string;
      34             : using std::map;
      35             : using std::vector;
      36             : using boost::shared_ptr;
      37             : using namespace boost::assign;
      38             : using std::pair;
      39             : using boost::system::error_code;
      40             : using process::ConnectionState;
      41             : using process::ConnectionType;
      42             : using process::ConnectionStatus;
      43             : using namespace boost::placeholders;
      44             : 
      45             : 
      46             : std::string Collector::prog_name_;
      47             : std::string Collector::self_ip_;
      48             : 
      49             : bool Collector::task_policy_set_ = false;
      50             : const std::string Collector::kDbTask = "analytics::DbHandler";
      51             : const int Collector::kQSizeHighWaterMark = 7 * 1024 * 1024;
      52             : const int Collector::kQSizeLowWaterMark  = 3 * 1024 * 1024;
      53             : 
      54             : const std::vector<Sandesh::QueueWaterMarkInfo> Collector::kDbQueueWaterMarkInfo =
      55             :     boost::assign::tuple_list_of
      56             :         (Collector::kQSizeHighWaterMark, SandeshLevel::INVALID, true, true)
      57             :         (Collector::kQSizeLowWaterMark, SandeshLevel::INVALID, false, true);
      58             : const std::vector<Sandesh::QueueWaterMarkInfo> Collector::kSmQueueWaterMarkInfo =
      59             :     boost::assign::tuple_list_of
      60             :         (Collector::kQSizeHighWaterMark, SandeshLevel::INVALID, true, true)
      61             :         (Collector::kQSizeLowWaterMark, SandeshLevel::INVALID, false, true);
      62             : 
      63           0 : Collector::Collector(EventManager *evm, short server_port,
      64             :         const std::string &server_ip,
      65             :         const SandeshConfig &sandesh_config, DbHandlerPtr db_handler,
      66           0 :         OpServerProxy *osp, VizCallback cb) :
      67             :         SandeshServer(evm, sandesh_config),
      68           0 :         db_handler_(db_handler),
      69           0 :         osp_(osp),
      70           0 :         evm_(evm),
      71           0 :         cb_(cb),
      72           0 :         db_queue_wm_info_(kDbQueueWaterMarkInfo),
      73           0 :         sm_queue_wm_info_(kSmQueueWaterMarkInfo) {
      74           0 :     SandeshServer::Initialize(server_port, server_ip);
      75             : 
      76           0 :     Module::type module = Module::COLLECTOR;
      77           0 :     string module_name = g_vns_constants.ModuleNames.find(module)->second;
      78           0 :     Sandesh::RecordPort("collector", module_name, GetPort());
      79           0 : }
      80             : 
      81           0 : Collector::~Collector() {
      82           0 : }
      83             : 
      84           0 : void Collector::SessionShutdown() {
      85           0 :     SandeshServer::SessionShutdown();
      86             : 
      87           0 :     std::scoped_lock lock(gen_map_mutex_);
      88           0 :     gen_map_.clear();
      89           0 : }
      90             : 
      91           0 : void Collector::Shutdown() {
      92           0 :     SandeshServer::Shutdown();
      93           0 : }
      94             : 
      95           0 : void Collector::RedisUpdate(bool rsc) {
      96           0 :     LOG(INFO, "RedisUpdate " << rsc);
      97             : 
      98           0 :     std::scoped_lock lock(gen_map_mutex_);
      99           0 :     for (GeneratorMap::iterator gen_it = gen_map_.begin();
     100           0 :             gen_it != gen_map_.end(); gen_it++) {
     101           0 :         SandeshGenerator *gen = gen_it->second;
     102           0 :         if (gen->session()) gen->get_state_machine()->ResourceUpdate(rsc);
     103             :     }
     104           0 :     return;
     105           0 : }
     106             : 
     107           0 : bool Collector::ReceiveResourceUpdate(SandeshSession *session,
     108             :             bool rsc) {
     109           0 :     VizSession *vsession = dynamic_cast<VizSession *>(session);
     110           0 :     if (!vsession) {
     111           0 :         increment_no_session_error();
     112           0 :         LOG(ERROR, __func__ << ": NO VizSession");
     113           0 :         return false;
     114             :     }
     115           0 :     SandeshGenerator *gen = vsession->generator();
     116           0 :     if (gen) {
     117           0 :         if (!rsc) {
     118           0 :             LOG(ERROR, "Force gen " << gen->ToString() <<
     119             :                 " to disconnect on redis disconnection");
     120           0 :             gen->DisconnectSession(vsession);
     121           0 :             return false;
     122             :         }
     123             : 
     124           0 :         std::vector<UVETypeInfo> vu;
     125           0 :         std::map<std::string, int32_t> seqReply;
     126           0 :         bool retc = osp_->GetSeq(gen->source(), gen->node_type(),
     127             :                         gen->module(), gen->instance_id(), seqReply);
     128           0 :         if (retc) {
     129           0 :             for (map<string,int32_t>::const_iterator it = seqReply.begin();
     130           0 :                     it != seqReply.end(); it++) {
     131           0 :                 UVETypeInfo uti;
     132           0 :                 uti.set_type_name(it->first);
     133           0 :                 uti.set_seq_num(it->second);
     134           0 :                 vu.push_back(uti);
     135           0 :             }
     136           0 :             SandeshCtrlServerToClient::Request(vu, retc, "ctrl", vsession->connection());
     137             :         } else {
     138           0 :             increment_redis_error();
     139           0 :             LOG(ERROR, "Resource OSP GetSeq FAILED: " << gen->ToString() <<
     140             :                 " Session: " << vsession->ToString());
     141           0 :             gen->DisconnectSession(vsession);
     142           0 :             return false;
     143             :         }
     144             : 
     145           0 :         return true;
     146           0 :     } else {
     147           0 :         increment_no_generator_error();
     148           0 :         LOG(ERROR, __func__ << "Resource State " << rsc <<
     149             :                 ": SandeshGenerator NOT PRESENT: Session: " << vsession->ToString());
     150           0 :         return false;
     151             :     }
     152             : }
     153             : 
     154           0 : bool Collector::ReceiveSandeshMsg(SandeshSession *session,
     155             :                                   const SandeshMessage *msg, bool rsc) {
     156           0 :     boost::uuids::uuid unm(umn_gen_());
     157             : 
     158           0 :     VizMsg vmsg(msg, unm);
     159             : 
     160           0 :     VizSession *vsession = dynamic_cast<VizSession *>(session);
     161           0 :     if (!vsession) {
     162           0 :         increment_no_session_error();
     163           0 :         LOG(ERROR, __func__ << ": NO VizSession");
     164           0 :         return false;
     165             :     }
     166           0 :     SandeshGenerator *gen = vsession->generator();
     167           0 :     if (gen) {
     168           0 :         return gen->ReceiveSandeshMsg(&vmsg, rsc);
     169             :     } else {
     170           0 :         increment_no_generator_error();
     171           0 :         LOG(ERROR, __func__ << ": Sandesh message " << msg->GetMessageType() <<
     172             :                 ": SandeshGenerator NOT PRESENT: Session: " << vsession->ToString());
     173           0 :         return false;
     174             :     }
     175           0 : }
     176             : 
     177           0 : SslSession* Collector::AllocSession(SslSocket *socket) {
     178           0 :     VizSession *session = new VizSession(this, socket, AllocConnectionIndex(),
     179           0 :                                          session_writer_task_id(),
     180           0 :                                          session_reader_task_id());
     181           0 :     return session;
     182             : }
     183             : 
     184           0 : bool Collector::ReceiveSandeshCtrlMsg(SandeshStateMachine *state_machine,
     185             :         SandeshSession *session, const Sandesh *sandesh) {
     186           0 :     VizSession *vsession = dynamic_cast<VizSession *>(session);
     187           0 :     if (!vsession) {
     188           0 :         increment_no_session_error();
     189           0 :         LOG(ERROR, "Received Ctrl Message without session " <<
     190             :                 sandesh->Name());
     191           0 :         return false;
     192             :     }
     193           0 :     assert(sandesh);
     194             :     const SandeshCtrlClientToServer *snh =
     195           0 :             dynamic_cast<const SandeshCtrlClientToServer *>(sandesh);
     196           0 :     if (!snh) {
     197           0 :         increment_sandesh_type_mismatch_error();
     198           0 :         LOG(ERROR, "Received Ctrl Message with wrong type " <<
     199             :                 sandesh->Name() << ": Session: " << vsession->ToString());
     200           0 :         return false;
     201             :     }
     202           0 :     if (snh->get_instance_id_name().empty()) {
     203           0 :         LOG(ERROR, "Received Ctrl Message with empty instance id from " <<
     204             :             snh->get_source() << ":" << snh->get_module_name());
     205           0 :         return false;
     206             :     }
     207           0 :     if (snh->get_node_type_name().empty()) {
     208           0 :         LOG(ERROR, "Received Ctrl Message with empty node type from " <<
     209             :             snh->get_source() << ":" << snh->get_module_name());
     210           0 :         return false;
     211             :     }
     212           0 :     if (!osp_->IsRedisInitDone()) {
     213           0 :         LOG(ERROR, "Collector connection to redis is not establised and flush done");
     214           0 :         return false;
     215             :     }
     216           0 :     SandeshGenerator::GeneratorId id(boost::make_tuple(snh->get_source(),
     217           0 :             snh->get_module_name(), snh->get_instance_id_name(),
     218           0 :             snh->get_node_type_name()));
     219             :     SandeshGenerator *gen;
     220             :     {
     221             :         // locked block with mutex gen_map_mutex_
     222           0 :         std::scoped_lock lock(gen_map_mutex_);
     223           0 :         GeneratorMap::iterator gen_it = gen_map_.find(id);
     224           0 :         if (gen_it == gen_map_.end()) {
     225           0 :             gen = new SandeshGenerator(this, vsession, state_machine, id.get<0>(),
     226           0 :                     id.get<1>(), id.get<2>(), id.get<3>(), db_handler_);
     227           0 :             gen_map_.insert(id, gen);
     228             :         } else {
     229             :             // Update the generator if needed
     230           0 :             gen = gen_it->second;
     231           0 :             VizSession *gsession = gen->session();
     232           0 :             if (gsession == NULL) {
     233           0 :                 gen->ConnectSession(vsession, state_machine);
     234             :             } else {
     235           0 :                 increment_session_mismatch_error();
     236             :                 // Message received on different session. Close both.
     237           0 :                 LOG(DEBUG, "Received Ctrl Message: " << gen->ToString()
     238             :                     << " On Session:" << vsession->ToString() <<
     239             :                     " Current Session:" << gsession->ToString());
     240             :                 // Enqueue a close on the state machine on the generator session
     241           0 :                 gsession->EnqueueClose();
     242           0 :                 return false;
     243             :             }
     244             :         }
     245           0 :         LOG(DEBUG, "Received Ctrl Message: " << gen->ToString()
     246             :                 << " Session:" << vsession->ToString());
     247           0 :         vsession->set_generator(gen);
     248           0 :     }
     249             : 
     250           0 :     std::vector<UVETypeInfo> vu;
     251           0 :     std::map<std::string, int32_t> seqReply;
     252           0 :     bool retc = osp_->GetSeq(snh->get_source(), snh->get_node_type_name(),
     253             :                              snh->get_module_name(), snh->get_instance_id_name(),
     254             :                              seqReply);
     255           0 :     if (retc) {
     256           0 :         for (map<string,int32_t>::const_iterator it = seqReply.begin();
     257           0 :              it != seqReply.end(); it++) {
     258           0 :             UVETypeInfo uti;
     259           0 :             uti.set_type_name(it->first);
     260           0 :             uti.set_seq_num(0);
     261           0 :             vu.push_back(uti);
     262           0 :         }
     263           0 :         SandeshCtrlServerToClient::Request(vu, retc, "ctrl", vsession->connection());
     264             :     } else {
     265           0 :         increment_redis_error();
     266           0 :         LOG(ERROR, "OSP GetSeq FAILED: " << gen->ToString() <<
     267             :             " Session:" << vsession->ToString());
     268           0 :         gen->DisconnectSession(vsession);
     269           0 :         return false;
     270             :     }
     271             : 
     272           0 :     LOG(DEBUG, "Sent good Ctrl Msg: Size " << vu.size() << " " <<
     273             :             snh->get_source() << ":" << snh->get_module_name() << ":" <<
     274             :             snh->get_instance_id_name() << ":" << snh->get_node_type_name());
     275           0 :     gen->ReceiveSandeshCtrlMsg(snh->get_sucessful_connections());
     276           0 :     return true;
     277           0 : }
     278             : 
     279           0 : void Collector::DisconnectSession(SandeshSession *session) {
     280           0 :     VizSession *vsession = dynamic_cast<VizSession *>(session);
     281           0 :     if (!vsession) {
     282           0 :         increment_no_session_error();
     283           0 :         LOG(ERROR, __func__ << " NO VizSession");
     284           0 :         return;
     285             :     }
     286           0 :     SandeshGenerator *gen = vsession->generator();
     287           0 :     assert(gen);
     288           0 :     LOG(INFO, "Received Disconnect: " << gen->ToString() << " Session:"
     289             :             << vsession->ToString());
     290           0 :     gen->DisconnectSession(vsession);
     291             : }
     292             : 
     293           0 : const std::string Collector::DbGlobalName(bool dup,
     294             :     const std::string &host_ip) {
     295           0 :     std::string name;
     296           0 :     error_code error;
     297           0 :     if (dup)
     298           0 :         name = ResolveCanonicalName(host_ip) + "dup" + ":" + "Global";
     299             :     else
     300           0 :         name = ResolveCanonicalName(host_ip) + ":" + "Global";
     301             : 
     302           0 :     return name;
     303           0 : }
     304             : 
     305           0 : void Collector::SendGeneratorStatistics() {
     306           0 :     std::scoped_lock lock(gen_map_mutex_);
     307           0 :     for (GeneratorMap::iterator gm_it = gen_map_.begin();
     308           0 :             gm_it != gen_map_.end(); gm_it++) {
     309           0 :         SandeshGenerator *gen = gm_it->second;
     310             :         // Only send if generator is connected
     311           0 :         VizSession *session = gen->session();
     312           0 :         if (!session) {
     313           0 :             continue;
     314             :         }
     315             :         // Sandesh message info
     316           0 :         gen->SendSandeshMessageStatistics();
     317             :     }
     318           0 : }
     319             : 
     320           0 : void Collector::GetGeneratorUVEInfo(vector<ModuleServerState> &genlist) {
     321           0 :     genlist.clear();
     322           0 :     std::scoped_lock lock(gen_map_mutex_);
     323           0 :     for (GeneratorMap::const_iterator gm_it = gen_map_.begin();
     324           0 :             gm_it != gen_map_.end(); gm_it++) {
     325           0 :         const SandeshGenerator * const gen = gm_it->second;
     326             : 
     327           0 :         vector<SandeshStats> ssv;
     328           0 :         gen->GetStatistics(&ssv);
     329           0 :         vector<SandeshLogLevelStats> lsv;
     330           0 :         gen->GetStatistics(&lsv);
     331           0 :         vector<SandeshStatsInfo> ssiv;
     332           0 :         SandeshStatsInfo ssi;
     333           0 :         ssi.set_hostname(Sandesh::source());
     334           0 :         ssi.set_msgtype_stats(ssv);
     335           0 :         ssi.set_log_level_stats(lsv);
     336           0 :         ssiv.push_back(ssi);
     337             : 
     338           0 :         ModuleServerState ginfo;
     339             :         uint64_t sm_queue_count;
     340           0 :         if (gen->GetSandeshStateMachineQueueCount(sm_queue_count)) {
     341           0 :             ginfo.set_sm_queue_count(sm_queue_count);
     342             :         }
     343           0 :         std::string sm_drop_level;
     344           0 :         if (gen->GetSandeshStateMachineDropLevel(sm_drop_level)) {
     345           0 :             ginfo.set_sm_drop_level(sm_drop_level);
     346             :         }
     347           0 :         SandeshStateMachineStats sm_stats;
     348           0 :         SandeshGeneratorBasicStats sm_msg_stats;
     349           0 :         if (gen->GetSandeshStateMachineStats(sm_stats, sm_msg_stats)) {
     350           0 :             ginfo.set_sm_stats(sm_stats);
     351           0 :             ginfo.set_sm_msg_stats(sm_msg_stats);
     352             :         }
     353             :         // Only send if generator is connected
     354           0 :         VizSession *session = gen->session();
     355           0 :         if (session) {
     356           0 :             ginfo.set_session_stats(session->GetStats());
     357           0 :             SocketIOStats rx_stats;
     358           0 :             session->GetRxSocketStats(&rx_stats);
     359           0 :             ginfo.set_session_rx_socket_stats(rx_stats);
     360           0 :             SocketIOStats tx_stats;
     361           0 :             session->GetTxSocketStats(&tx_stats);
     362           0 :             ginfo.set_session_tx_socket_stats(tx_stats);
     363           0 :         }
     364             : 
     365           0 :         ginfo.set_msg_stats(ssiv);
     366           0 :         ginfo.set_name(gen->ToString());
     367           0 :         genlist.push_back(ginfo);
     368           0 :     }
     369           0 : }
     370             : 
     371           0 : void Collector::GetGeneratorSummaryInfo(vector<GeneratorSummaryInfo> *genlist) {
     372           0 :     genlist->clear();
     373           0 :     std::scoped_lock lock(gen_map_mutex_);
     374           0 :     for (GeneratorMap::const_iterator gm_it = gen_map_.begin();
     375           0 :             gm_it != gen_map_.end(); gm_it++) {
     376           0 :         GeneratorSummaryInfo gsinfo;
     377           0 :         const SandeshGenerator * const gen = gm_it->second;
     378           0 :         ModuleServerState ginfo;
     379           0 :         gen->GetGeneratorInfo(ginfo);
     380           0 :         vector<GeneratorInfo> giv = ginfo.get_generator_info();
     381           0 :         GeneratorInfoAttr gen_attr = giv[0].get_gen_attr();
     382           0 :         if (gen_attr.get_connects() > gen_attr.get_resets()) {
     383           0 :             gsinfo.set_source(gm_it->first.get<0>());
     384           0 :             gsinfo.set_module_id(gm_it->first.get<1>());
     385           0 :             gsinfo.set_instance_id(gm_it->first.get<2>());
     386           0 :             gsinfo.set_node_type(gm_it->first.get<3>());
     387           0 :             gsinfo.set_state(gen->State());
     388             :             uint64_t sm_queue_count;
     389           0 :             if (gen->GetSandeshStateMachineQueueCount(sm_queue_count)) {
     390           0 :                 gsinfo.set_sm_queue_count(sm_queue_count);
     391             :             }
     392           0 :             gsinfo.set_sm_defer(
     393           0 :                 gen->IsStateMachineDeferTimerRunning());
     394           0 :             gsinfo.set_sm_defer_time_msec(
     395             :                 gen->GetStateMachineDeferTimeMSec());
     396           0 :             genlist->push_back(gsinfo);
     397             :         }
     398           0 :     }
     399           0 : }
     400             : 
     401           0 : bool Collector::SendRemote(const string& destination, const string& dec_sandesh) {
     402           0 :     std::vector<std::string> dest;
     403             :     // destination is of the format "source:module:instance_id:node_type"
     404             :     // source/module/instance_id/node_type can be wildcard
     405           0 :     boost::split(dest, destination, boost::is_any_of(":"),
     406             :                  boost::token_compress_on);
     407           0 :     if (dest.size() != 4) {
     408           0 :         LOG(ERROR, "Invalid destination " << destination << "." <<
     409             :             "Failed to send sandesh request: " << dec_sandesh);
     410           0 :         return false;
     411             :     }
     412           0 :     std::scoped_lock lock(gen_map_mutex_);
     413           0 :     for (GeneratorMap::const_iterator gm_it = gen_map_.begin();
     414           0 :             gm_it != gen_map_.end(); gm_it++) {
     415           0 :         SandeshGenerator::GeneratorId id(gm_it->first);
     416             :         // GeneratorId is of the format source..module..instance_id..node_type
     417           0 :         if (((dest[0] != "*") && (id.get<0>() != dest[0])) ||
     418           0 :             ((dest[1] != "*") && (id.get<3>() != dest[1])) ||
     419           0 :             ((dest[2] != "*") && (id.get<1>() != dest[2])) ||
     420           0 :             ((dest[3] != "*") && (id.get<2>() != dest[3]))) {
     421           0 :             continue;
     422             :         }
     423           0 :         const SandeshGenerator *gen = gm_it->second;
     424           0 :         SandeshSession *session = gen->session();
     425           0 :         if (session) {
     426           0 :             session->EnqueueBuffer((uint8_t *)dec_sandesh.c_str(), dec_sandesh.size());
     427             :         } else {
     428           0 :             increment_no_session_error();
     429           0 :             LOG(ERROR, "No connection to " << destination <<
     430             :                 ". Failed to send sandesh " << dec_sandesh);
     431             :         }
     432           0 :     }
     433           0 :     return true;
     434           0 : }
     435             : 
     436           0 : void Collector::SetQueueWaterMarkInfo(QueueType::type type,
     437             :     Sandesh::QueueWaterMarkInfo &wm) {
     438           0 :     std::scoped_lock lock(gen_map_mutex_);
     439           0 :     GeneratorMap::iterator gen_it = gen_map_.begin();
     440           0 :     for (; gen_it != gen_map_.end(); gen_it++) {
     441           0 :         SandeshGenerator *gen = gen_it->second;
     442           0 :         if (type == QueueType::Db) {
     443           0 :             gen->SetDbQueueWaterMarkInfo(wm);
     444           0 :         } else if (type == QueueType::Sm) {
     445           0 :             gen->SetSmQueueWaterMarkInfo(wm);
     446             :         }
     447             :     }
     448           0 :     if (type == QueueType::Db) {
     449           0 :         db_queue_wm_info_.push_back(wm);
     450           0 :     } else if (type == QueueType::Sm) {
     451           0 :         sm_queue_wm_info_.push_back(wm);
     452             :     }
     453           0 : }
     454             : 
     455           0 : void Collector::SetDbQueueWaterMarkInfo(Sandesh::QueueWaterMarkInfo &wm) {
     456           0 :     SetQueueWaterMarkInfo(QueueType::Db, wm);
     457           0 : }
     458             :     
     459           0 : void Collector::SetSmQueueWaterMarkInfo(Sandesh::QueueWaterMarkInfo &wm) {
     460           0 :     SetQueueWaterMarkInfo(QueueType::Sm, wm);
     461           0 : }
     462             : 
     463           0 : void Collector::ResetQueueWaterMarkInfo(QueueType::type type) {
     464           0 :     std::scoped_lock lock(gen_map_mutex_);
     465           0 :     GeneratorMap::iterator gen_it = gen_map_.begin();
     466           0 :     for (; gen_it != gen_map_.end(); gen_it++) {
     467           0 :         SandeshGenerator *gen = gen_it->second;
     468           0 :         if (type == QueueType::Db) {
     469           0 :             gen->ResetDbQueueWaterMarkInfo();
     470           0 :         } else if (type == QueueType::Sm) {
     471           0 :             gen->ResetSmQueueWaterMarkInfo();
     472             :         } 
     473             :     }
     474           0 :     if (type == QueueType::Db) {
     475           0 :         db_queue_wm_info_.clear();
     476           0 :     } else if (type == QueueType::Sm) {
     477           0 :         sm_queue_wm_info_.clear();
     478             :     }
     479           0 : }
     480             : 
     481           0 : void Collector::ResetDbQueueWaterMarkInfo() {
     482           0 :     ResetQueueWaterMarkInfo(QueueType::Db);
     483           0 : }
     484             :     
     485           0 : void Collector::ResetSmQueueWaterMarkInfo() {
     486           0 :     ResetQueueWaterMarkInfo(QueueType::Sm);
     487           0 : }
     488             : 
     489           0 : void Collector::GetQueueWaterMarkInfo(QueueType::type type,
     490             :     std::vector<Sandesh::QueueWaterMarkInfo> &wm_info) const {
     491           0 :     if (type == QueueType::Db) {
     492           0 :         wm_info = db_queue_wm_info_;
     493           0 :     } else if (type == QueueType::Sm) {
     494           0 :         wm_info = sm_queue_wm_info_;
     495             :     }
     496           0 : }
     497             : 
     498           0 : void Collector::GetDbQueueWaterMarkInfo(
     499             :     std::vector<Sandesh::QueueWaterMarkInfo> &wm_info) const {
     500           0 :     GetQueueWaterMarkInfo(QueueType::Db, wm_info);
     501           0 : }
     502             : 
     503           0 : void Collector::GetSmQueueWaterMarkInfo(
     504             :     std::vector<Sandesh::QueueWaterMarkInfo> &wm_info) const {
     505           0 :     GetQueueWaterMarkInfo(QueueType::Sm, wm_info);
     506           0 : }
     507             : 
     508           0 : void Collector::CloseGeneratorSession(string source, string module,
     509             :                          string instance, string node_type) {
     510             :     SandeshGenerator::GeneratorId id(boost::make_tuple(source,
     511           0 :                               module, instance, node_type));
     512           0 :     std::scoped_lock lock(gen_map_mutex_);
     513           0 :     GeneratorMap::iterator gen_it = gen_map_.find(id);
     514           0 :     if (gen_it != gen_map_.end()) {
     515           0 :         SandeshGenerator *gen = gen_it->second;
     516           0 :         VizSession *gsession = gen->session();
     517           0 :         if (gsession) {
     518           0 :             gsession->EnqueueClose();
     519             :         }
     520             :     }
     521           0 : }
     522             : 
     523           0 : DbHandlerPtr Collector::GetDbHandlerPtr() {
     524           0 :     return db_handler_;
     525             : }