Line data    Source code

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