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