Line data    Source code

       1             : /*
       2             :  * Copyright (c) 2017 Juniper Networks, Inc. All rights reserved.
       3             :  */
       4             : 
       5             : #include "config_amqp_client.h"
       6             : 
       7             : #include <boost/algorithm/string/find.hpp>
       8             : #include <boost/lexical_cast.hpp>
       9             : #include <stdio.h>
      10             : #include <string>
      11             : 
      12             : #include <SimpleAmqpClient/SimpleAmqpClient.h>
      13             : #include "rapidjson/document.h"
      14             : 
      15             : #include "base/connection_info.h"
      16             : #include "base/task.h"
      17             : #include "base/address_util.h"
      18             : #include "base/string_util.h"
      19             : #include "config_factory.h"
      20             : #include "config_cassandra_client.h"
      21             : #include "config_client_log.h"
      22             : #include "config_client_log_types.h"
      23             : #include "config_client_manager.h"
      24             : #include "config_db_client.h"
      25             : #include "config_client_show_types.h"
      26             : 
      27             : using namespace boost;
      28             : using namespace std;
      29             : using namespace contrail_rapidjson;
      30             : 
      31             : bool ConfigAmqpClient::disable_;
      32             : 
      33             : class ConfigAmqpClient::RabbitMQReader : public Task {
      34             : public:
      35           2 :     RabbitMQReader(ConfigAmqpClient *amqpclient) :
      36           2 :             Task(amqpclient->reader_task_id()), amqpclient_(amqpclient) {
      37           2 :         channel_.reset(ConfigStaticObjectFactory::Create<ConfigAmqpChannel>());
      38             : 
      39             :         // Connect to rabbit-mq asap so that notification messages over
      40             :         // rabbit mq are never missed (during bulk db sync which happens
      41             :         // soon afterwards.
      42           2 :         ConnectToRabbitMQ();
      43           2 :     }
      44             : 
      45             :     virtual bool Run();
      46           0 :     string Description() const { return "ConfigAmqpClient::RabbitMQReader"; }
      47             : 
      48             : private:
      49             :     ConfigAmqpClient *amqpclient_;
      50             :     boost::scoped_ptr<ConfigAmqpChannel> channel_;
      51             :     string consumer_tag_;
      52             :     void ConnectToRabbitMQ(bool queue_delete = true);
      53             :     bool AckRabbitMessages(AmqpClient::Envelope::ptr_t &envelop);
      54             :     bool ReceiveRabbitMessages(AmqpClient::Envelope::ptr_t &envelop);
      55             : };
      56             : 
      57        3673 : ConfigAmqpClient::ConfigAmqpClient(ConfigClientManager *mgr, string hostname,
      58        3673 :                       string module_name, const ConfigClientOptions &options) :
      59        3673 :     mgr_(mgr), hostname_(hostname), module_name_(module_name),
      60        3673 :     current_server_index_(0), terminate_(false),
      61        3673 :     rabbitmq_user_(options.rabbitmq_user),
      62        3673 :     rabbitmq_password_(options.rabbitmq_password),
      63        3673 :     rabbitmq_vhost_(options.rabbitmq_vhost),
      64        3673 :     rabbitmq_use_ssl_(options.rabbitmq_use_ssl),
      65        3673 :     rabbitmq_ssl_version_(options.rabbitmq_ssl_version),
      66        3673 :     rabbitmq_ssl_keyfile_(options.rabbitmq_ssl_keyfile),
      67        3673 :     rabbitmq_ssl_certfile_(options.rabbitmq_ssl_certfile),
      68       11019 :     rabbitmq_ssl_ca_certs_(options.rabbitmq_ssl_ca_certs) {
      69             : 
      70        3673 :     connection_status_ = false;
      71        3673 :     connection_status_change_at_ = UTCTimestampUsec();
      72             : 
      73        3673 :     TaskScheduler *scheduler = TaskScheduler::GetInstance();
      74        3673 :     reader_task_id_ = scheduler->GetTaskId("amqp::RabbitMQReader");
      75             : 
      76        3673 :     if (options.rabbitmq_server_list.empty())
      77        3671 :         return;
      78             : 
      79           2 :     for (vector<string>::const_iterator iter =
      80           2 :                 options.rabbitmq_server_list.begin();
      81           4 :          iter != options.rabbitmq_server_list.end(); iter++) {
      82           2 :         string server_info(*iter);
      83             :         typedef boost::tokenizer<boost::char_separator<char> > tokenizer;
      84           2 :         boost::char_separator<char> sep(":");
      85           2 :         tokenizer tokens(server_info, sep);
      86           2 :         tokenizer::iterator tit = tokens.begin();
      87           2 :         string ip(*tit);
      88           2 :         rabbitmq_ips_.push_back(ip);
      89           2 :         ++tit;
      90           2 :         string port_str(*tit);
      91           2 :         rabbitmq_ports_.push_back(port_str);
      92           2 :         Endpoint curr_ep;
      93           2 :         int port = 0;
      94           2 :         stringToInteger(port_str, port);
      95           2 :         boost::system::error_code ec;
      96           2 :         curr_ep.address(AddressFromString(ip, &ec));
      97           2 :         curr_ep.port(port);
      98           2 :         endpoints_.push_back(curr_ep);
      99           2 :     }
     100           0 : }
     101             : 
     102        7740 : void ConfigAmqpClient::StartRabbitMQReader() {
     103        7740 :     if (disable_)  {
     104        7738 :         CONFIG_CLIENT_DEBUG(
     105             :             ConfigClientMgrDebug,
     106             :             "RabbitMQ SM: StartRabbitMQReader: RabbitMQ disabled");
     107        7738 :         return;
     108             :     }
     109             : 
     110             :     // If reinit is triggerred, Don't start the rabbitmq reader
     111           2 :     if (config_manager()->is_reinit_triggered()) {
     112           0 :         CONFIG_CLIENT_DEBUG(
     113             :             ConfigClientMgrDebug,
     114             :             "RabbitMQ SM: StartRabbitMQReader: re init triggered,"
     115             :             " dont start RabbitMQ");
     116           0 :         return;
     117             :     }
     118           2 :     TaskScheduler *scheduler = TaskScheduler::GetInstance();
     119           2 :     Task *task = new RabbitMQReader(this);
     120           2 :     scheduler->Enqueue(task);
     121             : }
     122             : 
     123       44492 : void ConfigAmqpClient::EnqueueUUIDRequest(string oper, string obj_type,
     124             :                                      string uuid_str) {
     125       44492 :     if (mgr_->config_json_parser()->IsReadObjectType(obj_type)) {
     126       44492 :         mgr_->EnqueueUUIDRequest(oper, obj_type, uuid_str);
     127             :     }
     128       44492 : }
     129             : 
     130         328 : string ConfigAmqpClient::FormAmqpUri(bool hide_auth_info) const {
     131         328 :     const string user = hide_auth_info ? "********" : rabbitmq_user();
     132         328 :     const string password = hide_auth_info ? "********" : rabbitmq_password();
     133         656 :     string uri = string("amqp://" + user + ":" +
     134        1312 :                         password + "@" + rabbitmq_ip() + ":" + rabbitmq_port());
     135         328 :     if (!rabbitmq_vhost().empty()) {
     136         106 :         if (rabbitmq_vhost().compare("/") != 0) {
     137         106 :             uri += "/" + rabbitmq_vhost();
     138             :         }
     139             :     }
     140         656 :     return uri;
     141         328 : }
     142             : 
     143         208 : void ConfigAmqpClient::ReportRabbitMQConnectionStatus(bool connected) const {
     144         208 :     if (connected) {
     145             :         // Update connection info
     146         312 :         process::ConnectionState::GetInstance()->Update(
     147             :             process::ConnectionType::DATABASE, "RabbitMQ",
     148             :             process::ConnectionStatus::UP,
     149         208 :             endpoints(), "RabbitMQ connection established");
     150         104 :         CONFIG_CLIENT_DEBUG(ConfigClientMgrDebug,
     151             :                             "RabbitMQ SM: RabbitMQ connection established");
     152             :     } else {
     153         312 :         process::ConnectionState::GetInstance()->Update(
     154             :             process::ConnectionType::DATABASE, "RabbitMQ",
     155             :             process::ConnectionStatus::DOWN,
     156         208 :             endpoints(), "RabbitMQ connection down");
     157         104 :         CONFIG_CLIENT_DEBUG(ConfigClientMgrDebug,
     158             :                             "RabbitMQ SM: RabbitMQ connection down");
     159             :     }
     160         208 : }
     161             : 
     162         104 : void ConfigAmqpClient::RabbitMQReader::ConnectToRabbitMQ(bool queue_delete) {
     163         104 :     amqpclient_->ReportRabbitMQConnectionStatus(false);
     164         104 :     amqpclient_->set_connected(false);
     165         104 :     string message = "RabbitMQ SM: Connect to Rabbit MQ with queue_delete ";
     166         104 :     message += queue_delete ? "TRUE" : "FALSE";
     167         104 :     CONFIG_CLIENT_DEBUG(ConfigClientMgrDebug, message);
     168         104 :     size_t count = 0;
     169             :     while (true) {
     170             :         // If we are signalled to stop, break now.
     171         104 :         if (amqpclient_->config_manager()->is_reinit_triggered()) {
     172           0 :             CONFIG_CLIENT_DEBUG(ConfigClientMgrDebug,
     173             :                                 "RabbitMQ SM: Skipped connect due to reinit");
     174           0 :             return;
     175             :         }
     176         104 :         string uri = amqpclient_->FormAmqpUri(false);
     177             :         try {
     178         104 :             if (amqpclient_->rabbitmq_use_ssl()) {
     179           0 :                 int port = boost::lexical_cast<int>(
     180           0 :                         amqpclient_->rabbitmq_port());
     181             : 
     182           0 :                 channel_->CreateSecure(
     183           0 :                     amqpclient_->rabbitmq_ssl_ca_certs(),
     184           0 :                     amqpclient_->rabbitmq_ip(),
     185           0 :                     amqpclient_->rabbitmq_ssl_keyfile(),
     186           0 :                     amqpclient_->rabbitmq_ssl_certfile(),
     187             :                     port,
     188           0 :                     amqpclient_->rabbitmq_user(),
     189           0 :                     amqpclient_->rabbitmq_password(),
     190           0 :                     amqpclient_->rabbitmq_vhost());
     191             :             } else {
     192         104 :                 channel_->CreateFromUri(uri);
     193             :             }
     194             :             // passive = false, durable = false, auto_delete = false
     195         104 :             channel_->DeclareExchange("vnc_config.object-update",
     196             :               AmqpClient::Channel::EXCHANGE_TYPE_FANOUT, false, false, false);
     197             :             string queue_name =
     198         208 :                 amqpclient_->module_name() + "." + amqpclient_->hostname();
     199             : 
     200         104 :             if (queue_delete) {
     201           2 :                 channel_->DeleteQueue(queue_name, false, false);
     202             :             }
     203             : 
     204             :             // passive = false, durable = false,
     205             :             // exclusive = false, auto_delete = false
     206         104 :             string queue = channel_->DeclareQueue(queue_name, false, false,
     207         104 :                                                   false, false);
     208         104 :             channel_->BindQueue(queue, "vnc_config.object-update");
     209             :             // no_local = true, no_ack = false,
     210             :             // exclusive = false, message_prefetch_count = 0
     211         208 :             consumer_tag_ = channel_->BasicConsume(queue, queue_name,
     212         104 :                                                    true, false, false, 0);
     213         104 :         } catch (std::exception &e) {
     214           0 :             static string what = e.what();
     215             :             string message =
     216             :                 "RabbitMQ SM: Caught exception while connecting to RabbitMQ: "
     217           0 :                 + amqpclient_->rabbitmq_ip() + ":"
     218           0 :                 + amqpclient_->rabbitmq_port() + " : " + what;
     219           0 :             cout << message << endl;
     220           0 :             CONFIG_CLIENT_WARN(ConfigClientMgrWarning, message);
     221           0 :             if (++count == amqpclient_->rabbitmq_server_list_len()) {
     222           0 :                 count = 0;
     223             :                 // Tried connecting to all given servers.. Now wait to reconnect
     224           0 :                 sleep(5);
     225             :             }
     226           0 :             amqpclient_->increment_rabbitmq_server_index();
     227           0 :             continue;
     228           0 :         } catch (...) {
     229             :             string message =
     230             :                 "RabbitMQ SM: Caught fatal exception while "
     231             :                 "connecting to RabbitMQ: "
     232           0 :                 + amqpclient_->rabbitmq_ip() + ":"
     233           0 :                 + amqpclient_->rabbitmq_port();
     234           0 :             cout << message << endl;
     235           0 :             CONFIG_CLIENT_WARN(ConfigClientMgrWarning, message);
     236           0 :             assert(0);
     237           0 :         }
     238             : 
     239         104 :         amqpclient_->ReportRabbitMQConnectionStatus(true);
     240         104 :         amqpclient_->set_connected(true);
     241         104 :         break;
     242         104 :     }
     243         104 : }
     244             : 
     245         208 : void ConfigAmqpClient::set_connected(bool connected) {
     246         208 :     connection_status_ = connected;
     247         208 :     connection_status_change_at_ = UTCTimestampUsec();
     248         208 : }
     249             : 
     250         224 : void ConfigAmqpClient::GetConnectionInfo(ConfigAmqpConnInfo &conn_info) const {
     251         224 :     conn_info.connection_status = connection_status_;
     252             :     conn_info.connection_status_change_at =
     253         224 :         UTCUsecToString(connection_status_change_at_);
     254         224 :     conn_info.url = FormAmqpUri(true);
     255         224 : }
     256             : 
     257         574 : bool ConfigAmqpClient::ProcessMessage(const string &json_message) {
     258         574 :     Document document;
     259         574 :     document.Parse<0>(json_message.c_str());
     260             : 
     261         574 :     if (document.HasParseError()) {
     262           2 :         size_t pos = document.GetErrorOffset();
     263             :         // GetParseError returns const char *
     264           2 :         cout << "Error in parsing JSON message from rabbitMQ at "
     265           2 :             << pos << "with error description"
     266           2 :             << document.GetParseError() << endl;
     267           2 :         return false;
     268             :     } else {
     269         572 :         string oper = "";
     270         572 :         string uuid_str = "";
     271         572 :         string obj_type = "";
     272         572 :         string obj_name = "";
     273         572 :         for (Value::ConstMemberIterator itr = document.MemberBegin();
     274        2860 :              itr != document.MemberEnd(); ++itr) {
     275        2288 :             string key(itr->name.GetString());
     276        2288 :             if (key == "oper") {
     277         572 :                 oper = itr->value.GetString();
     278        1716 :             } else if (key == "type") {
     279         572 :                 obj_type = itr->value.GetString();
     280        1144 :             } else if (key == "fq_name") {
     281         572 :                 if (!itr->value.IsArray())
     282          64 :                     continue;
     283         508 :                 ostringstream os;
     284         508 :                 SizeType sz = itr->value.GetArray().Size();
     285         508 :                 if (sz == 0)
     286           0 :                     continue;
     287        1002 :                 for (SizeType i = 0; i < sz-1; i++) {
     288         494 :                     os << itr->value[i].GetString() << ":";
     289             :                 }
     290         508 :                 os << itr->value[sz-1].GetString();
     291         508 :                 obj_name = os.str();
     292        1080 :             } else if (key == "uuid") {
     293         572 :                 uuid_str = itr->value.GetString();
     294             :             }
     295        2288 :         }
     296             : 
     297         572 :         if ((oper == "") || (uuid_str == "") || (obj_type == "")) {
     298           0 :             CONFIG_CLIENT_WARN(ConfigClientFQNameCache,
     299             :                 "Empty object name or empty type or empty uuid", obj_type,
     300             :                 obj_name, uuid_str);
     301           0 :             return false;
     302             :         }
     303             : 
     304         572 :         if ((oper == "CREATE") || (oper == "UPDATE")) {
     305         432 :             if (obj_name.empty()) {
     306           0 :                 CONFIG_CLIENT_WARN(ConfigClientFQNameCache,
     307             :                     "Empty object name during CREATE/UPDATE",
     308             :                     obj_type, obj_name, uuid_str);
     309           0 :                 return false;
     310             :             }
     311             : 
     312             :             // It is possible in some cases that RabbitMQ might club the 
     313             :             // CREATE and UPDATE for an object if they happen in quick 
     314             :             // succession (for instance control node failover). In such 
     315             :             // cases, we could only get an UPDATE of the object without 
     316             :             // a preceding CREATE. Handle the UPDATE as a CREATE and add
     317             :             //  object to FQNameCache.
     318             :             // In the unlikely event that we do receive a CREATE for the
     319             :             // same object after the UPDATE, we check if the FQNameCache
     320             :             // is already present before adding to it.
     321             :             // Also, it is ok to process the CREATE/UPDATE irrespective
     322             :             // of the order in which they are received since we always
     323             :             // read the uuid table from Cassandra.
     324             : 
     325             :             string stored_fq_name =
     326         432 :                 config_manager()->config_db_client()->FindFQName(uuid_str);
     327         432 :             if (stored_fq_name == "ERROR") {
     328             :                 // FQName Cache entry not present. Create one.
     329             :                 // Log the event if the operation is an UPDATE.
     330         432 :                 if (oper == "UPDATE") {
     331         160 :                     CONFIG_CLIENT_WARN(ConfigClientFQNameCache,
     332             :                             "FQ Name Cache entry not found on UPDATE:",
     333             :                             obj_type, obj_name, uuid_str);
     334             :                 }
     335         432 :                 config_manager()->config_db_client()->
     336         432 :                             AddFQNameCache(uuid_str, obj_type, obj_name);
     337             :             }
     338         572 :         } else if (oper == "DELETE") {
     339         140 :             config_manager()->config_db_client()->
     340         140 :                 InvalidateFQNameCache(uuid_str);
     341             :         }
     342             : 
     343         572 :         CONFIG_CLIENT_RABBIT_MSG_TRACE(ConfigClientRabbitMQMsgTrace, oper,
     344             :                                        obj_type, obj_name, uuid_str);
     345         572 :         EnqueueUUIDRequest(oper, obj_type, uuid_str);
     346         572 :     }
     347         572 :     return true;
     348         574 : }
     349             : 
     350         204 : bool ConfigAmqpClient::RabbitMQReader::ReceiveRabbitMessages(
     351             :                                      AmqpClient::Envelope::ptr_t &envelope) {
     352             :     try {
     353             :         // timeout = 10ms.. To handle SIGHUP on config changes
     354             :         // On reinit, config client manager will trigger the amqp client
     355             :         // to shutdown. Blocking wait without timeout will not allow this.
     356         204 :         channel_->BasicConsumeMessage(consumer_tag_, envelope, 10);
     357         102 :         return true;
     358         102 :     } catch (std::exception &e) {
     359         102 :         static string what = e.what();
     360             :         string message =
     361             :             "RabbitMQ SM: Caught exception while receiving "
     362             :             "messages from RabbitMQ: "
     363         204 :             + amqpclient_->rabbitmq_ip() + ":"
     364         306 :             + amqpclient_->rabbitmq_port() + " : " + what;
     365         102 :         cout << message << endl;
     366         102 :         CONFIG_CLIENT_WARN(ConfigClientMgrWarning, message);
     367         102 :         return false;
     368         102 :     } catch (...) {
     369             :         string message =
     370             :             "RabbitMQ SM: Caught fatal unknown exception while receiving "
     371             :             "messages from RabbitMQ "
     372           0 :             + amqpclient_->rabbitmq_ip() + ':'
     373           0 :             + amqpclient_->rabbitmq_port();
     374           0 :         cout << message << endl;
     375           0 :         CONFIG_CLIENT_WARN(ConfigClientMgrWarning, message);
     376           0 :         assert(0);
     377           0 :     }
     378             :     return true;
     379             : }
     380             : 
     381           0 : bool ConfigAmqpClient::RabbitMQReader::AckRabbitMessages(
     382             :                                      AmqpClient::Envelope::ptr_t &envelope) {
     383             :     try {
     384           0 :         channel_->BasicAck(envelope);
     385           0 :     } catch (std::exception &e) {
     386           0 :         static string what = e.what();
     387             :         string message =
     388             :             "RabbitMQ SM: Caught exception while acking "
     389             :             "messages from RabbitMQ: "
     390           0 :             + amqpclient_->rabbitmq_ip() + ':'
     391           0 :             + amqpclient_->rabbitmq_port() + ':' + what;
     392           0 :         cout << message << endl;
     393           0 :         CONFIG_CLIENT_WARN(ConfigClientMgrWarning, message);
     394           0 :         return false;
     395           0 :     } catch (...) {
     396             :         string message =
     397             :             "RabbitMQ SM: Caught fatal unknown exception while acking messages "
     398           0 :             "from RabbitMQ " + amqpclient_->rabbitmq_ip() + ':'
     399           0 :             + amqpclient_->rabbitmq_port();
     400           0 :         cout << message << endl;
     401           0 :         CONFIG_CLIENT_WARN(ConfigClientMgrWarning, message);
     402           0 :         assert(0);
     403           0 :     }
     404           0 :     return true;
     405             : }
     406             : 
     407             : 
     408           2 : bool ConfigAmqpClient::RabbitMQReader::Run() {
     409             :     // If reinit is triggerred, don't wait for end of config trigger
     410             :     // return from here to process reinit
     411           2 :     if (amqpclient_->config_manager()->is_reinit_triggered()) {
     412           0 :         CONFIG_CLIENT_DEBUG(
     413             :             ConfigClientMgrDebug,
     414             :             "RabbitMQ SM: Reinit triggered, don't wait for end of config");
     415           0 :         return true;
     416             :     }
     417             : 
     418             :     // To start consuming the message, we should have finised bulk sync
     419           2 :     amqpclient_->config_manager()->WaitForEndOfConfig();
     420             : 
     421             :     while (true) {
     422             :         // Test only
     423         206 :         if (amqpclient_->terminate())
     424           2 :             break;
     425             :         // If reinit is triggerred, break from the message receiving loop
     426         204 :         if (amqpclient_->config_manager()->is_reinit_triggered()) {
     427           0 :             CONFIG_CLIENT_DEBUG(ConfigClientMgrDebug,
     428             :             "RabbitMQ SM: Reinit triggered, break from message receiving loop");
     429           0 :             break;
     430             :         }
     431         204 :         AmqpClient::Envelope::ptr_t envelope;
     432         204 :         if (ReceiveRabbitMessages(envelope) == false) {
     433         102 :             ConnectToRabbitMQ(false);
     434         102 :             continue;
     435             :         }
     436             : 
     437         102 :         if (!envelope)
     438         102 :             continue;
     439             : 
     440           0 :         amqpclient_->ProcessMessage(envelope->Message()->Body());
     441           0 :         if (AckRabbitMessages(envelope) == false) {
     442           0 :             ConnectToRabbitMQ(false);
     443           0 :             continue;
     444             :         }
     445         408 :     }
     446           2 :     return true;
     447             : }