Line data    Source code

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