source/diameter.comm/Entity.cpp

   1 // ANNA - Anna is Not Nothingness Anymore                                                         //
   2 //                                                                                                //
   3 // (c) Copyright 2005-2015 Eduardo Ramos Testillano & Francisco Ruiz Rayo                         //
   4 //                                                                                                //
   5 // See project site at http://redmine.teslayout.com/projects/anna-suite                           //
   6 // See accompanying file LICENSE or copy at http://www.teslayout.com/projects/public/anna.LICENSE //
   7
   8
   9 #include <anna/diameter.comm/Server.hpp>
  10 #include <anna/diameter.comm/Entity.hpp>
  11 #include <anna/diameter.comm/Engine.hpp>
  12 #include <anna/diameter.comm/OamModule.hpp>
  13 #include <anna/diameter/codec/functions.hpp>
  14 #include <anna/diameter/helpers/base/defines.hpp>
  15 #include <anna/core/functions.hpp>
  16 #include <anna/time/functions.hpp>
  17 #include <anna/diameter/helpers/base/functions.hpp>
  18 #include <anna/diameter/helpers/dcca/functions.hpp>
  19
  20 #include <anna/core/tracing/Logger.hpp>
  21 #include <anna/core/functions.hpp>
  22 #include <anna/core/tracing/TraceMethod.hpp>
  23
  24
  25 using namespace anna;
  26 using namespace anna::diameter;
  27 using namespace anna::diameter::comm;
  28
  29
  30
  31 void Entity::initialize() throw() {
  32   a_engine = NULL;
  33   a_servers.clear(); // importante (el recycler creo que no lo tocaba)
  34   a_available = false;
  35   a_deprecated = false;
  36   a_socketListLiteral = "";
  37   a_primarySocketLiteral = "";
  38   a_secondarySocketLiteral = "";
  39   a_description = "";
  40   a_category = 0;
  41   a_lastUsedResource = NULL;
  42 }
  43
  44
  45 void Entity::assertReady() throw(anna::RuntimeException) {
  46   if(a_servers.size() != a_maxServers) {
  47     std::string msg(asString());
  48     msg += " | Non-configured entity: you must add the remaining servers before any operation (bind, send, etc.)";
  49     throw anna::RuntimeException(msg, ANNA_FILE_LOCATION);
  50   }
  51 }
  52
  53
  54 void Entity::addServer(const socket_t & serverId)
  55 throw(anna::RuntimeException) {
  56   if(a_servers.size() == a_maxServers) {
  57     LOGDEBUG
  58     (
  59       std::string msg = "Maximum number of servers reached for this entity (";
  60       msg += anna::functions::asString(a_maxServers);
  61       msg += "). Please use clear()/close() primitives";
  62       anna::Logger::debug(msg, ANNA_FILE_LOCATION);
  63     );
  64     return;
  65   }
  66
  67   if(!a_engine)
  68     throw anna::RuntimeException("Invalid engine reference (NULL)", ANNA_FILE_LOCATION);
  69
  70   Server *s = a_engine->createServer(this, serverId);
  71   a_servers.push_back(s);
  72   a_deliveryIterator = begin();
  73 }
  74
  75
  76 int Entity::readSocketId(const Message* message, int maxClientSessions) const throw() {
  77   try {
  78     // Service-Context-Id:
  79     anna::diameter::helpers::dcca::ChargingContext::_v chargingContext;
  80     std::string scid = anna::diameter::helpers::dcca::functions::getServiceContextId(message->getBody(), chargingContext);
  81
  82     switch(chargingContext) {
  83     case anna::diameter::helpers::dcca::ChargingContext::Data:
  84     case anna::diameter::helpers::dcca::ChargingContext::Voice:
  85     case anna::diameter::helpers::dcca::ChargingContext::Content: {
  86       // Session-Id: '<DiameterIdentity>;<high 32 bits>;<low 32 bits>[;<optional value>="">]'
  87       std::string sid = anna::diameter::helpers::base::functions::getSessionId(message->getBody());
  88       std::string diameterIdentity, optional;
  89       anna::U32 high, low;
  90       anna::diameter::helpers::base::functions::decodeSessionId(sid, diameterIdentity, high, low /* context-teid */, optional);
  91       return (low % maxClientSessions);
  92     }
  93     case anna::diameter::helpers::dcca::ChargingContext::SMS:
  94     case anna::diameter::helpers::dcca::ChargingContext::MMS:
  95     case anna::diameter::helpers::dcca::ChargingContext::Unknown:
  96     default:
  97        return -1; // IEC model and Unknown traffic types
  98     }
  99   } catch(anna::RuntimeException &ex) {
 100     LOGDEBUG(
 101       std::string msg = ex.getText();
 102       msg += " | Round-robin between sessions will be used to send";
 103       anna::Logger::debug(msg, ANNA_FILE_LOCATION);
 104     );
 105   }
 106
 107   return -1;
 108 }
 109
 110
 111 bool Entity::send(const Message* message, bool balance) throw(anna::RuntimeException) {
 112   LOGMETHOD(anna::TraceMethod tttm("diameter::comm::Entity", "send", ANNA_FILE_LOCATION));
 113   assertReady();
 114   // Carried socket id (forwarding/proxy features):
 115   std::string carriedSocketId = message->getRequestClientSessionKey();
 116
 117   if(carriedSocketId != "") {
 118     // Send (it was a request, we forward the answer):
 119     try {
 120       ClientSession * fixedClientSession = a_engine->findClientSession(carriedSocketId); // exception if not found
 121       fixedClientSession->send(message);
 122       return true;
 123     } catch(anna::RuntimeException &ex) {
 124       std::string msg = "Cannot deliver answer through a fixed client session (";
 125       msg += carriedSocketId;
 126       msg += "). Perhaps it existed but not now. Ignore";
 127       anna::Logger::error(msg, ANNA_FILE_LOCATION);
 128       ex.trace();
 129     }
 130   }
 131
 132   ////////////////////////////////////////////////////////////////////////////////////////
 133   // BALANCE vs STANDARD
 134   // Balance algorythm remember last delivery resource used, balancing from start to end,
 135   // understanding start as next resource to last used, and end as last used. Standard
 136   // algorithm always starts at primary (first defined) server.
 137   ////////////////////////////////////////////////////////////////////////////////////////
 138
 139   // Balance
 140   if(balance) {
 141     for(int k = 0; k < getMaxServers(); k++) {   // try round-robin only over one cycle,
 142       // no matter where you are: don't repeat same server
 143       if(a_deliveryIterator == end()) a_deliveryIterator = begin();
 144
 145       a_lastUsedResource = (*a_deliveryIterator);
 146       a_deliveryIterator++;
 147       // At 'readSocketId' documentation:
 148       //      If server is configured as single session (max client sessions equal to 1), entity will ignore
 149       //      this method because it won't affect the session selection.
 150       int serverSessions = a_lastUsedResource->getMaxClientSessions();
 151       int socketId = (serverSessions > 1) ? readSocketId(message, serverSessions) : -1; // optimization
 152
 153       if(a_lastUsedResource->send(message, socketId))  // exception only possible at findClientSession within server send procedure
 154         return true;
 155     }
 156   } else {
 157     // Standard (no balance) //   start at begining, try secondary, and so on until end.
 158     std::vector<Server*>::iterator it = begin();
 159
 160     while(it != end()) {
 161       a_lastUsedResource = (*it);
 162       it++;
 163       // At 'readSocketId' documentation:
 164       //      If server is configured as single session (max client sessions equal to 1), entity will ignore
 165       //      this method because it won't affect the session selection.
 166       int serverSessions = a_lastUsedResource->getMaxClientSessions();
 167       int socketId = (serverSessions > 1) ? readSocketId(message, serverSessions) : -1; // optimization
 168
 169       if(a_lastUsedResource->send(message, socketId))  // exception only possible at findClientSession within server send procedure
 170         return true;
 171     }
 172   }
 173
 174   // END BALANCE AND TRY ALGORITHM or STANDARD ///////////////////////////////////////////
 175   ////////////////////////////////////////////////////////////////////////////////////////
 176   // Here, sent has failed:
 177   // OAM
 178   OamModule &oamModule = OamModule::instantiate();
 179
 180   if(a_maxServers != 2) {
 181     oamModule.activateAlarm(OamModule::Alarm::UnableToDeliverDiameterMessageToEntityDefinedAs__s__, a_socketListLiteral.c_str());
 182     oamModule.count(OamModule::Counter::UnableToDeliverOverEntity);
 183   } else {
 184     OamModule &oamModule = OamModule::instantiate();
 185     oamModule.activateAlarm(OamModule::Alarm::UnableToDeliverDiameterMessageToEntityDefinedAsPrimary__s__AndSecondary__s__,
 186                             a_primarySocketLiteral.c_str(),
 187                             a_secondarySocketLiteral.c_str());
 188     oamModule.count(OamModule::Counter::UnableToDeliverOverEntity);
 189   }
 190
 191   return false;
 192 }
 193
 194 bool Entity::broadcast(const Message* message) throw(anna::RuntimeException) {
 195   LOGMETHOD(anna::TraceMethod tttm("diameter::comm::Entity", "broadcast", ANNA_FILE_LOCATION));
 196   assertReady();
 197   bool allok = true;
 198   bool ok;
 199
 200   for(std::vector<Server*>::iterator it = begin(); it != end(); it++) {
 201     try {
 202       ok = (*it)->broadcast(message);
 203
 204       if(!ok) allok = false;
 205     } catch(anna::RuntimeException &ex) {
 206       ex.trace();
 207       allok = false;
 208     }
 209   }
 210
 211   return allok;
 212 }
 213
 214 bool Entity::bind() throw(anna::RuntimeException) {
 215   LOGMETHOD(anna::TraceMethod tttm("diameter::comm::Entity", "bind", ANNA_FILE_LOCATION));
 216   assertReady();
 217   bool result = true; // all OK return
 218
 219   for(std::vector<Server*>::iterator it = begin(); it != end(); it++) {
 220     try {
 221       (*it)->bind();
 222     } catch(anna::RuntimeException &ex) {
 223       ex.trace();
 224       result = false;
 225     }
 226   }
 227
 228   return result;
 229 }
 230
 231 void Entity::raiseAutoRecovery(bool autoRecovery) throw(anna::RuntimeException) {
 232   LOGMETHOD(anna::TraceMethod tttm("diameter::comm::Entity", "raiseAutoRecovery", ANNA_FILE_LOCATION));
 233   assertReady();
 234
 235   for(std::vector<Server*>::iterator it = begin(); it != end(); it++)
 236     (*it)->raiseAutoRecovery(autoRecovery);
 237 }
 238
 239 void Entity::setClassCodeTimeout(const ClassCode::_v v, const anna::Millisecond & millisecond) throw() {
 240   LOGMETHOD(anna::TraceMethod tttm("diameter::comm::Entity", "setClassCodeTimeout", ANNA_FILE_LOCATION));
 241   assertReady();
 242
 243   for(std::vector<Server*>::iterator it = begin(); it != end(); it++) {
 244     try {
 245       (*it)->setClassCodeTimeout(v, millisecond);
 246     } catch(anna::RuntimeException &ex) {
 247       ex.trace();
 248     }
 249   }
 250 }
 251
 252
 253 // Private close/destroy method
 254 void Entity::close(bool destroy) throw(anna::RuntimeException) {
 255   LOGMETHOD(anna::TraceMethod tttm("diameter::comm::Entity", "close", ANNA_FILE_LOCATION));
 256
 257   if(!a_engine)
 258     throw anna::RuntimeException("Invalid engine reference (NULL)", ANNA_FILE_LOCATION);
 259
 260   assertReady();
 261
 262   for(std::vector<Server*>::iterator it = begin(); it != end(); it++)
 263     a_engine->closeServer(*it, destroy);
 264 }
 265
 266
 267 socket_v Entity::getAddressPortList() const throw() {
 268   socket_v result;
 269
 270   for(std::vector<Server*>::const_iterator it = begin(); it != end(); it++) {
 271     socket_t address((*it)->getAddress(), (*it)->getPort());
 272     result.push_back(address);
 273   }
 274
 275   return result;
 276 }
 277
 278 int Entity::getOTARequests() const throw() {
 279   int result = 0;
 280
 281   for(std::vector<Server*>::const_iterator it = begin(); it != end(); it++)
 282     result += (*it)->getOTARequests();
 283
 284   return result;
 285 }
 286
 287 void Entity::childIdle() const throw() {
 288   // Check father engine idleness:
 289   if(idle()) a_engine->eraseDeprecatedIdleEntities();
 290 }
 291
 292
 293 void Entity::hide() throw() {
 294   for(std::vector<Server*>::iterator it = begin(); it != end(); it++)
 295     (*it)->hide();
 296 }
 297
 298 void Entity::show() throw() {
 299   for(std::vector<Server*>::iterator it = begin(); it != end(); it++)
 300     (*it)->show();
 301 }
 302
 303 bool Entity::hidden() const throw() {
 304   for(std::vector<Server*>::const_iterator it = begin(); it != end(); it++)
 305     if((*it)->shown()) return false;
 306
 307   return true;
 308 }
 309 bool Entity::shown() const throw() {
 310   for(std::vector<Server*>::const_iterator it = begin(); it != end(); it++)
 311     if((*it)->hidden()) return false;
 312
 313   return true;
 314 }
 315
 316 void Entity::eventPeerShutdown(const ClientSession* clientSession) throw() {
 317   LOGWARNING(
 318     std::string msg(clientSession->asString());
 319     msg += " | eventPeerShutdown";
 320     anna::Logger::warning(msg, ANNA_FILE_LOCATION);
 321   );
 322 }
 323
 324 void Entity::eventRequestRetransmission(const ClientSession* clientSession, Message *request) throw() {
 325   LOGWARNING(
 326     std::string msg(clientSession->asString());
 327
 328     HopByHop hopByHop = codec::functions::getHopByHop(request->getBody()); // context identification
 329     int retries = request->getRetries();
 330
 331     msg += anna::functions::asString(" | eventRequestRetransmission: request with application HopByHop: %u; remaining %d retries", hopByHop, retries);
 332     anna::Logger::warning(msg, ANNA_FILE_LOCATION);
 333   );
 334 }
 335
 336 std::string Entity::asString() const throw() {
 337   std::string result("diameter::comm::Entity { ");
 338   std::string originRealm = a_engine->getOriginRealm();
 339   std::string originHost = a_engine->getOriginHost();
 340
 341   result += "Parent Engine Origin-Realm: ";
 342   result += (originRealm != "") ? originRealm:"<not configured>";
 343   result += " | Parent Engine Origin-Host: ";
 344   result += (originHost != "") ? originHost:"<not configured>";
 345
 346   result += " | Category: ";
 347   result += anna::functions::asString(a_category);
 348
 349   if(a_description != "") {
 350     result += " | Description: '";
 351     result += a_description;
 352   }
 353
 354   result += " | Available: ";
 355   result += a_available ? "yes" : "no";
 356   result += " | Deprecated: ";
 357   result += a_deprecated ? "yes" : "no";
 358   result += " | Max servers supported: ";
 359   result += anna::functions::asString(a_maxServers);
 360   result += " | Currently configured servers: ";
 361   result += anna::functions::asString(a_servers.size());
 362   result += anna::functions::asString(" | OTA requests: %d%s", getOTARequests(), idle() ? " (idle)" : "");
 363   result += " | Last Incoming Activity Time: ";
 364   result += a_lastIncomingActivityTime.asString();
 365   result += " | Last Outgoing Activity Time: ";
 366   result += a_lastOutgoingActivityTime.asString();
 367   result += " | Hidden: ";
 368   result += (hidden() ? "yes" : "no");
 369   result += "\n";
 370
 371   for(std::vector<Server*>::const_iterator it = begin(); it != end(); it++) {
 372     result += "\n";
 373     result += (*it)->asString();
 374   }
 375
 376   return result;
 377 }
 378
 379 anna::xml::Node* Entity::asXML(anna::xml::Node* parent) const throw() {
 380   anna::xml::Node* result = parent->createChild("diameter.Entity");
 381   std::string originRealm = a_engine->getOriginRealm();
 382   std::string originHost = a_engine->getOriginHost();
 383
 384   if(originRealm != "") result->createAttribute("ParentEngineOriginRealm", originRealm);
 385   if(originHost != "") result->createAttribute("ParentEngineOriginHost", originHost);
 386
 387   result->createAttribute("Category", anna::functions::asString(a_category));
 388
 389   if(a_description != "") result->createAttribute("Description", a_description);
 390
 391   result->createAttribute("Available", a_available ? "yes" : "no");
 392   result->createAttribute("Deprecated", a_deprecated ? "yes" : "no");
 393   result->createAttribute("MaxServersSupported", anna::functions::asString(a_maxServers));
 394   result->createAttribute("CurrentlyConfiguredServers", anna::functions::asString(a_servers.size()));
 395   result->createAttribute("OTArequests", anna::functions::asString("%d%s", getOTARequests(), idle() ? " (idle)" : ""));
 396   result->createAttribute("LastIncomingActivityTime", a_lastIncomingActivityTime.asString());
 397   result->createAttribute("LastOutgoingActivityTime", a_lastOutgoingActivityTime.asString());
 398   result->createAttribute("Hidden", hidden() ? "yes" : "no");
 399   anna::xml::Node* servers = result->createChild("Entity.Servers");
 400
 401   for(std::vector<Server*>::const_iterator it = begin(); it != end(); it++)
 402     (*it)->asXML(servers);
 403
 404   return result;
 405 }
 406
 407 void Entity::availabilityLost() throw() {
 408   a_available = false;
 409   LOGDEBUG(
 410     std::string msg = "diameter::comm::Entity { Description: ";
 411     msg += getDescription();
 412     msg += " } has lost its availability";
 413     anna::Logger::debug(msg, ANNA_FILE_LOCATION);
 414   );
 415   OamModule &oamModule = OamModule::instantiate();
 416
 417   if(a_maxServers != 2) {
 418     oamModule.activateAlarm(OamModule::Alarm::c_LostAvailabilityOverEntityDefinedAs__s__, a_socketListLiteral.c_str());
 419     oamModule.count(OamModule::Counter::LostAvailabilityOverEntity);
 420   } else {
 421     oamModule.activateAlarm(OamModule::Alarm::c_LostAvailabilityOverEntityDefinedAsPrimary__s__AndSecondary__s__,
 422                             a_primarySocketLiteral.c_str(),
 423                             a_secondarySocketLiteral.c_str());
 424     oamModule.count(OamModule::Counter::LostAvailabilityOverEntity);
 425   }
 426
 427   a_engine->availabilityLost(this);
 428   a_engine->refreshAvailabilityForEntities();
 429 }
 430
 431
 432 void Entity::availabilityRecovered() throw() {
 433   a_available = true;
 434   LOGDEBUG(
 435     std::string msg = "diameter::comm::Entity { Description: ";
 436     msg += getDescription();
 437     msg += " } has recovered its availability";
 438     anna::Logger::debug(msg, ANNA_FILE_LOCATION);
 439   );
 440   OamModule &oamModule = OamModule::instantiate();
 441
 442   if(a_maxServers != 2) {
 443     oamModule.cancelAlarm(OamModule::Alarm::c_LostAvailabilityOverEntityDefinedAs__s__, a_socketListLiteral.c_str());
 444     oamModule.count(OamModule::Counter::RecoveredAvailabilityOverEntity);
 445   } else {
 446     oamModule.cancelAlarm(OamModule::Alarm::c_LostAvailabilityOverEntityDefinedAsPrimary__s__AndSecondary__s__,
 447                           a_primarySocketLiteral.c_str(),
 448                           a_secondarySocketLiteral.c_str());
 449     oamModule.count(OamModule::Counter::RecoveredAvailabilityOverEntity);
 450   }
 451
 452   a_engine->availabilityRecovered(this);
 453   a_engine->refreshAvailabilityForEntities();
 454 }
 455
 456
 457 bool Entity::refreshAvailability() throw() {
 458   // Here available
 459   if(a_available) {  // check not-bound state for all servers:
 460     bool isolate = true;
 461
 462     for(std::vector<Server*>::const_iterator it = begin(); it != end(); it++)
 463       if((*it)->isAvailable()) { isolate = false; break; }
 464
 465     if(isolate) {
 466       availabilityLost();
 467       return true;
 468     }
 469
 470     return false;
 471   }
 472
 473   // Here not available
 474   for(std::vector<Server*>::const_iterator it = begin(); it != end(); it++)
 475     if((*it)->isAvailable()) {
 476       availabilityRecovered();
 477       return true;
 478     }
 479
 480   return false;
 481 }
 482
 483 //------------------------------------------------------------------------------
 484 //----------------------------------------- Entity::updateIncomingActivityTime()
 485 //------------------------------------------------------------------------------
 486 void Entity::updateIncomingActivityTime() throw() {
 487   a_lastIncomingActivityTime = anna::functions::millisecond();
 488   LOGDEBUG
 489   (
 490     std::string msg = "Updated INCOMING activity on entity (milliseconds unix): ";
 491     msg += anna::functions::asString(a_lastIncomingActivityTime.getValue());
 492     anna::Logger::debug(msg, ANNA_FILE_LOCATION);
 493   );
 494 }
 495
 496
 497 //------------------------------------------------------------------------------
 498 //----------------------------------------- Entity::updateOutgoingActivityTime()
 499 //------------------------------------------------------------------------------
 500 void Entity::updateOutgoingActivityTime(void) throw() {
 501   a_lastOutgoingActivityTime = anna::functions::millisecond();
 502   LOGDEBUG
 503   (
 504     std::string msg = "Updated OUTGOING activity on entity (milliseconds unix): ";
 505     msg += anna::functions::asString(a_lastOutgoingActivityTime.getValue());
 506     anna::Logger::debug(msg, ANNA_FILE_LOCATION);
 507   );
 508 }
 509