nelns/admin_service/admin_service.cpp

   1 // NeLNS - MMORPG Framework <http://dev.ryzom.com/projects/nel/>
   2 // Copyright (C) 2010  Winch Gate Property Limited
   3 //
   4 // This source file has been modified by the following contributors:
   5 // Copyright (C) 2019  Jan BOON (Kaetemi) <jan.boon@kaetemi.be>
   6 //
   7 // This program is free software: you can redistribute it and/or modify
   8 // it under the terms of the GNU Affero General Public License as
   9 // published by the Free Software Foundation, either version 3 of the
  10 // License, or (at your option) any later version.
  11 //
  12 // This program is distributed in the hope that it will be useful,
  13 // but WITHOUT ANY WARRANTY; without even the implied warranty of
  14 // MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  15 // GNU Affero General Public License for more details.
  16 //
  17 // You should have received a copy of the GNU Affero General Public License
  18 // along with this program.  If not, see <http://www.gnu.org/licenses/>.
  19
  20 #ifdef HAVE_CONFIG_H
  21 #include "config.h"
  22 #endif // HAVE_CONFIG_H
  23
  24 #ifndef NELNS_CONFIG
  25 #define NELNS_CONFIG ""
  26 #endif // NELNS_CONFIG
  27
  28 #ifndef NELNS_LOGS
  29 #define NELNS_LOGS ""
  30 #endif // NELNS_LOGS
  31
  32 #include "nel/misc/types_nl.h"
  33
  34 #include <string>
  35 #include <list>
  36
  37 #ifdef NL_OS_WINDOWS
  38 #include <winsock2.h>
  39 #include <windows.h>
  40 typedef unsigned long ulong;
  41 #endif
  42
  43 #include <mysql.h>
  44 #ifndef LIBMARIADB
  45 #include <mysql_version.h>
  46 #endif
  47
  48 #include "nel/misc/debug.h"
  49 #include "nel/misc/config_file.h"
  50 #include "nel/misc/path.h"
  51 #include "nel/misc/command.h"
  52
  53 #include "nel/net/service.h"
  54 #include "nel/net/varpath.h"
  55 #include "nel/net/email.h"
  56
  57 #include "connection_web.h"
  58
  59
  60 //
  61 // Namespaces
  62 //
  63
  64 using namespace std;
  65 using namespace NLMISC;
  66 using namespace NLNET;
  67
  68
  69 //
  70 // NeL Variables (for config file, etc)
  71 //
  72
  73 // this variable should be used in conjunction with UseExplicitAESRegistration.
  74 // the AS / AES registration process works as follows:
  75 // - aes creates a layer 5 connection to as
  76 // - as gets a serviceUp callback and looks in the database to try to find a match for the AES
  77 // - if the match fails then AS sends a reject message to the AES
  78 // - when the AES receives the reject message they check their UseExplicitAESRegistration flag - if it's set they
  79 //   attempt an explicit connection, sending the info required by the AS that would normally come from the database
  80 // - when the AS receives an explicit registration, it verifies the state of the AllowExplicitAESRegistration flag
  81 //   and completes the registration work that failed earlier due to the database access failure
  82 CVariable<bool> AllowExplicitAESRegistration("as","AllowExplicitAESRegistration","flag to allow AES services to register explicitly",false,0,true);
  83
  84 // this variable allows one to launch an AS on a machine that doesn't have a database setup
  85 // the functionality of the AS is reduced (particularly in respect to alarms and graphs which are configured via the database)
  86 CVariable<bool> DontUseDataBase("as","DontUseDataBase","if this flag is set calls to the database will be ignored",false,0,true);
  87
  88
  89 //
  90 // Structures
  91 //
  92
  93 struct CRequest
  94 {
  95         CRequest (uint32 id, TSockId from) : Id(id), NbWaiting(0), NbReceived(0), From(from), NbRow(0), NbLines(1)
  96         {
  97                 Time = CTime::getSecondsSince1970 ();
  98         }
  99
 100         uint32                  Id;
 101         uint                    NbWaiting;
 102         uint32                  NbReceived;
 103         TSockId                 From;
 104         uint32                  Time;   // when the request was ask
 105
 106         uint32                  NbRow;
 107         uint32                  NbLines;
 108
 109         vector<vector<string> > Array;  // it's the 2 dimensional array that will be send to the php for variables
 110         vector<string> Log;                             // this log contains the answer if a command was asked, othewise, Array contains the results
 111
 112         uint32 getVariable(const string &variable)
 113         {
 114                 for (uint32 i = 0; i < NbRow; i++)
 115                         if (Array[i][0] == variable)
 116                                 return i;
 117
 118                 // need to add the variable
 119                 vector<string> NewRow;
 120                 NewRow.resize (NbLines);
 121                 NewRow[0] = variable;
 122                 Array.push_back (NewRow);
 123                 return NbRow++;
 124         }
 125
 126         void addLine ()
 127         {
 128                 for (uint32 i = 0; i < NbRow; i++)
 129                         Array[i].push_back("");
 130
 131                 NbLines++;
 132         }
 133
 134         void display ()
 135         {
 136                 if (Log.empty())
 137                 {
 138                         nlinfo ("Display answer array for request %d: %d row %d lines", Id, NbRow, NbLines);
 139                         for (uint i = 0; i < NbLines; i++)
 140                         {
 141                                 for (uint j = 0; j < NbRow; j++)
 142                                 {
 143                                         nlassert (Array.size () == NbRow);
 144                                         InfoLog->displayRaw ("%-20s", Array[j][i].c_str());
 145                                 }
 146                                 InfoLog->displayRawNL ("");
 147                         }
 148                         InfoLog->displayRawNL ("End of the array");
 149                 }
 150                 else
 151                 {
 152                         nlinfo ("Display the log for request %d: %d lines", Id, Log.size());
 153                         for (uint i = 0; i < Log.size(); i++)
 154                         {
 155                                 InfoLog->displayRaw ("%s", Log[i].c_str());
 156                         }
 157                         InfoLog->displayRawNL ("End of the log");
 158                 }
 159         }
 160 };
 161
 162 struct CAdminExecutorService
 163 {
 164         CAdminExecutorService (const string &shard, const string &name, TServiceId sid) : Shard(shard), SId(sid), Name(name) { }
 165
 166         string  Shard;                  /// Name of the shard
 167         TServiceId      SId;                    /// uniq number to identify the AES
 168         string  Name;                   /// name of the admin executor service
 169
 170         vector<uint32>  WaitingRequestId;               /// contains all request that the server hasn't reply yet
 171
 172 };
 173
 174 typedef list<CAdminExecutorService> TAdminExecutorServices;
 175 typedef list<CAdminExecutorService>::iterator AESIT;
 176
 177
 178 //
 179 // Variables
 180 //
 181
 182 TAdminExecutorServices AdminExecutorServices;
 183
 184 MYSQL *DatabaseConnection = NULL;
 185
 186 vector<CRequest> Requests;
 187
 188 uint32 RequestTimeout = 5;      // in second
 189
 190 // cumulate 5 seconds of alert
 191 sint32 AdminAlertAccumlationTime = 5;
 192
 193
 194 //
 195 // Functions
 196 //
 197
 198 AESIT findAES (TServiceId sid, bool asrt = true)
 199 {
 200         AESIT aesit;
 201         for (aesit = AdminExecutorServices.begin(); aesit != AdminExecutorServices.end(); aesit++)
 202                 if ((*aesit).SId == sid)
 203                         break;
 204
 205                 if (asrt)
 206                         nlassert (aesit != AdminExecutorServices.end());
 207                 return aesit;
 208 }
 209
 210 AESIT findAES (const string &name, bool asrt = true)
 211 {
 212         AESIT aesit;
 213         for (aesit = AdminExecutorServices.begin(); aesit != AdminExecutorServices.end(); aesit++)
 214                 if ((*aesit).Name == name)
 215                         break;
 216
 217                 if (asrt)
 218                         nlassert (aesit != AdminExecutorServices.end());
 219
 220                 return aesit;
 221 }
 222
 223
 224 //
 225 // SQL helpers
 226 //
 227
 228 MYSQL_RES *sqlCurrentQueryResult = NULL;
 229
 230 MYSQL_ROW sqlQuery (const char *format, ...)
 231 {
 232         if (DontUseDataBase)
 233                 return 0;
 234
 235         char *query;
 236         NLMISC_CONVERT_VARGS (query, format, 1024);
 237
 238         if (DatabaseConnection == 0)
 239         {
 240                 nlwarning ("MYSQL: mysql_query (%s) failed: DatabaseConnection is 0", query);
 241                 return NULL;
 242         }
 243
 244         int ret = mysql_query (DatabaseConnection, query);
 245         if (ret != 0)
 246         {
 247                 nlwarning ("MYSQL: mysql_query () failed for query '%s': %s", query,  mysql_error(DatabaseConnection));
 248                 return 0;
 249         }
 250
 251         sqlCurrentQueryResult = mysql_store_result(DatabaseConnection);
 252         if (sqlCurrentQueryResult == 0)
 253         {
 254                 nlwarning ("MYSQL: mysql_store_result () failed for query '%s': %s", query,  mysql_error(DatabaseConnection));
 255                 return 0;
 256         }
 257
 258         MYSQL_ROW row = mysql_fetch_row(sqlCurrentQueryResult);
 259         if (row == 0)
 260         {
 261                 nlwarning ("MYSQL: mysql_fetch_row () failed for query '%s': %s", query,  mysql_error(DatabaseConnection));
 262         }
 263
 264         nldebug ("MYSQL: sqlQuery(%s) returns %d rows", query, mysql_num_rows(sqlCurrentQueryResult));
 265
 266         return row;
 267 }
 268
 269 MYSQL_ROW sqlNextRow ()
 270 {
 271         if (DontUseDataBase)
 272                 return 0;
 273
 274         if (sqlCurrentQueryResult == 0)
 275                 return 0;
 276
 277         return mysql_fetch_row(sqlCurrentQueryResult);
 278 }
 279
 280 void    sqlFlushResult()
 281 {
 282         if (DontUseDataBase)
 283                 return;
 284
 285         if (sqlCurrentQueryResult == NULL)
 286                 return;
 287
 288         mysql_free_result(sqlCurrentQueryResult);
 289         sqlCurrentQueryResult = NULL;
 290 }
 291
 292
 293 //
 294 // Admin functions
 295 //
 296
 297 string Email;
 298 uint32 FirstEmailTime = 0;
 299
 300 void sendAdminAlert (const char *format, ...)
 301 {
 302         char *text;
 303         NLMISC_CONVERT_VARGS (text, format, 4096);
 304
 305         if (AdminAlertAccumlationTime == -1)
 306         {
 307                 // we don't send email so just display a warning
 308                 nlwarning ("ALERT: %s", text);
 309         }
 310         else
 311         {
 312                 if(Email.empty() && FirstEmailTime == 0)
 313                 {
 314                         Email += text;
 315                         FirstEmailTime = CTime::getSecondsSince1970();
 316                 }
 317                 else
 318                 {
 319                         Email += "\n";
 320                         Email += text;
 321                 }
 322                 nldebug ("ALERT: pushing email into queue: %s", text);
 323         }
 324 }
 325
 326 void updateSendAdminAlert ()
 327 {
 328         if(!Email.empty() && FirstEmailTime != 0 && AdminAlertAccumlationTime >=0 && CTime::getSecondsSince1970() > FirstEmailTime + AdminAlertAccumlationTime)
 329         {
 330                 vector<string> lines;
 331                 explode (Email, string("\n"), lines, true);
 332
 333                 if (!lines.empty())
 334                 {
 335
 336                         if (IService::getInstance()->ConfigFile.exists("SysLogPath") && IService::getInstance()->ConfigFile.exists("SysLogParams"))
 337                         {
 338                                 // syslog
 339                                 string param;
 340                                 if (lines.size() > 1)
 341                                 {
 342                                         param = "Multiple problems, first is: ";
 343                                 }
 344                                 param += lines[0];
 345                                 string res = toString(IService::getInstance()->ConfigFile.getVar("SysLogParams").asString().c_str(), param.c_str());
 346                                 launchProgram(IService::getInstance()->ConfigFile.getVar("SysLogPath").asString(), res);
 347                         }
 348
 349                         if (IService::getInstance()->ConfigFile.exists("AdminEmail"))
 350                         {
 351                                 // email
 352                                 string subject;
 353                                 if (lines.size() == 1)
 354                                 {
 355                                         subject = lines[0];
 356                                 }
 357                                 else
 358                                 {
 359                                         subject = "Multiple problems";
 360                                 }
 361
 362                                 std::string from;
 363                                 if(IService::getInstance()->ConfigFile.exists("AdminEmailFrom"))
 364                                         from = IService::getInstance()->ConfigFile.getVar("AdminEmailFrom").asString();
 365                                 CConfigFile::CVar &var = IService::getInstance()->ConfigFile.getVar("AdminEmail");
 366                                 for (uint i = 0; i < var.size(); i++)
 367                                 {
 368                                         if (!sendEmail ("", from, var.asString(i), subject, Email))
 369                                         {
 370                                                 nlwarning ("Can't send email to '%s'", var.asString(i).c_str());
 371                                         }
 372                                         else
 373                                         {
 374                                                 nlinfo ("ALERT: Sent email to admin %s the subject: %s", var.asString(i).c_str(), subject.c_str());
 375                                         }
 376                                 }
 377                         }
 378                 }
 379
 380                 Email = "";
 381                 FirstEmailTime = 0;
 382         }
 383 }
 384
 385
 386 static void cbAdminEmail (CMessage &msgin, const std::string &serviceName, TServiceId sid)
 387 {
 388         string str;
 389         msgin.serial(str);
 390         sendAdminAlert (str.c_str());
 391 }
 392
 393 static void cbGraphUpdate (CMessage &msgin, const std::string &serviceName, TServiceId sid)
 394 {
 395         uint32 CurrentTime;
 396         msgin.serial (CurrentTime);
 397
 398         while (msgin.getPos() < (sint32)msgin.length())
 399         {
 400                 string var, service;
 401                 sint32 val;
 402                 msgin.serial (service, var, val);
 403
 404                 AESIT aesit = findAES (sid);
 405
 406                 string shard, server;
 407                 shard = (*aesit).Shard;
 408                 server = (*aesit).Name;
 409
 410                 if (!shard.empty() && !server.empty() && !service.empty() && !var.empty())
 411                 {
 412                         string path = CPath::standardizePath (IService::getInstance()->ConfigFile.getVar("RRDVarPath").asString());
 413                         string rrdfilename = path + shard+"."+server+"."+service+"."+var+".rrd";
 414
 415                         string arg;
 416
 417                         if (!NLMISC::CFile::fileExists(rrdfilename))
 418                         {
 419                                 MYSQL_ROW row = sqlQuery ("select graph_update from variable where path like '%%%s' and graph_update!=0", var.c_str());
 420                                 if (row != NULL)
 421                                 {
 422                                         uint32 freq = atoi(row[0]);
 423                                         arg = "create "+rrdfilename+" --step "+toString(freq)+" DS:var:GAUGE:"+toString(freq*2)+":U:U RRA:AVERAGE:0.5:1:1000 RRA:AVERAGE:0.5:10:1000 RRA:AVERAGE:0.5:100:1000";
 424                                         launchProgram(IService::getInstance()->ConfigFile.getVar("RRDToolPath").asString(), arg);
 425                                 }
 426                                 else
 427                                 {
 428                                         nlwarning ("Can't create the rrd because no graph_update in database");
 429                                 }
 430                                 sqlFlushResult();
 431                         }
 432
 433                         arg = "update " + rrdfilename + " " + toString (CurrentTime) + ":" + toString(val);
 434                         launchProgram(IService::getInstance()->ConfigFile.getVar("RRDToolPath").asString(), arg);
 435                 }
 436                 else
 437                 {
 438                         nlwarning ("Shard server service var val is empty");
 439                 }
 440         }
 441 }
 442
 443
 444 //
 445 // Request functions
 446 //
 447
 448 uint32 newRequest (TSockId from)
 449 {
 450         static uint32 NextId = 5461231;
 451
 452         Requests.push_back (CRequest(NextId, from));
 453
 454         return NextId++;
 455 }
 456
 457 void addRequestWaitingNb (uint32 rid)
 458 {
 459         for (uint i = 0 ; i < Requests.size (); i++)
 460         {
 461                 if (Requests[i].Id == rid)
 462                 {
 463                         Requests[i].NbWaiting++;
 464                         Requests[i].Time = CTime::getSecondsSince1970 ();
 465                         return;
 466                 }
 467         }
 468         nlwarning ("REQUEST: Received an answer from an unknown resquest %d (perhaps due to a AS timeout)", rid);
 469 }
 470
 471 void subRequestWaitingNb (uint32 rid)
 472 {
 473         for (uint i = 0 ; i < Requests.size (); i++)
 474         {
 475                 if (Requests[i].Id == rid)
 476                 {
 477                         Requests[i].NbWaiting--;
 478                         return;
 479                 }
 480         }
 481         nlwarning ("REQUEST: Received an answer from an unknown resquest %d (perhaps due to a AS timeout)", rid);
 482 }
 483
 484 void addRequestReceived (uint32 rid)
 485 {
 486         for (uint i = 0 ; i < Requests.size (); i++)
 487         {
 488                 if (Requests[i].Id == rid)
 489                 {
 490                         Requests[i].NbReceived++;
 491                         return;
 492                 }
 493         }
 494         nlwarning ("REQUEST: Received an answer from an unknown resquest %d (perhaps due to a AS timeout)", rid);
 495 }
 496
 497 void addRequestAnswer (uint32 rid, const vector<string> &variables, const vector<string> &values)
 498 {
 499         for (uint i = 0 ; i < Requests.size (); i++)
 500         {
 501                 Requests[i].addLine ();
 502                 if (Requests[i].Id == rid)
 503                 {
 504                         if (!variables.empty() && variables[0]=="__log")
 505                         {
 506                                 nlassert (variables.size() == 1);
 507
 508                                 for (uint j = 0; j < values.size(); j++)
 509                                 {
 510                                         Requests[i].Log.push_back (values[j]);
 511                                 }
 512                         }
 513                         else
 514                         {
 515                                 nlassert (variables.size() == values.size ());
 516                                 for (uint j = 0; j < variables.size(); j++)
 517                                 {
 518                                         uint32 pos = Requests[i].getVariable (variables[j]);
 519                                         Requests[i].Array[pos][Requests[i].NbLines-1] = values[j];
 520                                 }
 521                         }
 522                         return;
 523                 }
 524         }
 525         nlwarning ("REQUEST: Received an answer from an unknown resquest %d (perhaps due to a AS timeout)", rid);
 526 }
 527
 528 bool emptyRequest (uint32 rid)
 529 {
 530         for (uint i = 0 ; i < Requests.size (); i++)
 531         {
 532                 if (Requests[i].Id == rid && Requests[i].NbWaiting != 0)
 533                 {
 534                         return false;
 535                 }
 536         }
 537         return true;
 538 }
 539
 540 void cleanRequest ()
 541 {
 542         uint32 currentTime = CTime::getSecondsSince1970 ();
 543
 544         bool timeout;
 545
 546         for (uint i = 0 ; i < Requests.size ();)
 547         {
 548                 // the AES doesn't answer quickly
 549                 timeout = (currentTime >= Requests[i].Time+RequestTimeout);
 550
 551                 if (Requests[i].NbWaiting <= Requests[i].NbReceived || timeout)
 552                 {
 553                         // the request is over, send to the php
 554
 555                         string str;
 556
 557                         if (timeout)
 558                         {
 559                                 nlwarning ("REQUEST: Request %d timeouted, only %d on %d services have replied", Requests[i].Id, Requests[i].NbReceived, Requests[i].NbWaiting);
 560                         }
 561
 562                         if (Requests[i].Log.empty())
 563                         {
 564                                 if (Requests[i].NbRow == 0 && timeout)
 565                                 {
 566                                         str = "1 ((TIMEOUT))";
 567                                 }
 568                                 else
 569                                 {
 570                                         str = toString(Requests[i].NbRow) + " ";
 571                                         for (uint k = 0; k < Requests[i].NbLines; k++)
 572                                         {
 573                                                 for (uint j = 0; j < Requests[i].NbRow; j++)
 574                                                 {
 575                                                         nlassert (Requests[i].Array.size () == Requests[i].NbRow);
 576                                                         if (Requests[i].Array[j][k].empty ())
 577                                                                 str += "??? ";
 578                                                         else
 579                                                         {
 580                                                                 str += Requests[i].Array[j][k];
 581                                                                 if (timeout)
 582                                                                         str += "((TIMEOUT))";
 583                                                                 str += " ";
 584                                                         }
 585                                                 }
 586                                         }
 587                                 }
 588                         }
 589                         else
 590                         {
 591                                 for (uint k = 0; k < Requests[i].Log.size(); k++)
 592                                 {
 593                                         str += Requests[i].Log[k];
 594                                         if (timeout)
 595                                                 str += "((TIMEOUT))";
 596                                 }
 597                         }
 598
 599                         sendString (Requests[i].From, str);
 600
 601                         // set to 0 to erase it
 602                         Requests[i].NbWaiting = 0;
 603                 }
 604
 605                 if (Requests[i].NbWaiting == 0)
 606                 {
 607                         Requests.erase (Requests.begin ()+i);
 608                 }
 609                 else
 610                 {
 611                         i++;
 612                 }
 613         }
 614 }
 615
 616
 617 //
 618 // SQL functions
 619 //
 620
 621 void sqlInit ()
 622 {
 623         if (DontUseDataBase)
 624                 return;
 625
 626         MYSQL *db = mysql_init(NULL);
 627         if(db == NULL)
 628         {
 629                 nlerror ("mysql_init() failed");
 630         }
 631
 632         my_bool opt = true;
 633         if (mysql_options (db, MYSQL_OPT_RECONNECT, &opt))
 634         {
 635                 mysql_close(db);
 636                 DatabaseConnection = 0;
 637                 nlerror("mysql_options() failed for database connection to '%s'", IService::getInstance()->ConfigFile.getVar("DatabaseHost").asString().c_str());
 638                 return;
 639         }
 640
 641         DatabaseConnection = mysql_real_connect(db,
 642                 IService::getInstance()->ConfigFile.getVar("DatabaseHost").asString().c_str(),
 643                 IService::getInstance()->ConfigFile.getVar("DatabaseLogin").asString().c_str(),
 644                 IService::getInstance()->ConfigFile.getVar("DatabasePassword").asString().c_str(),
 645                 IService::getInstance()->ConfigFile.getVar("DatabaseName").asString().c_str(),
 646                 0,NULL,0);
 647         if (DatabaseConnection == NULL || DatabaseConnection != db)
 648         {
 649                 nlerror ("mysql_real_connect() failed to '%s' with login '%s' and database name '%s' with %s",
 650                         IService::getInstance()->ConfigFile.getVar("DatabaseHost").asString().c_str(),
 651                         IService::getInstance()->ConfigFile.getVar("DatabaseLogin").asString().c_str(),
 652                         IService::getInstance()->ConfigFile.getVar("DatabaseName").asString().c_str(),
 653                         (IService::getInstance()->ConfigFile.getVar("DatabasePassword").asString().empty()?"empty password":"password")
 654                         );
 655         }
 656
 657 #if MYSQL_VERSION_ID < 50019
 658         opt = true;
 659         if (mysql_options (DatabaseConnection, MYSQL_OPT_RECONNECT, &opt))
 660         {
 661                 mysql_close(db);
 662                 DatabaseConnection = 0;
 663                 nlerror("mysql_options() failed for database connection to '%s'", IService::getInstance()->ConfigFile.getVar("DatabaseHost").asString().c_str());
 664                 return;
 665         }
 666 #endif
 667 }
 668
 669
 670 ////////////////////////////////////////////////////////////////////////////////////////////////////////
 671 ////////////////////////////////////////////////////////////////////////////////////////////////////////
 672 ////////////////// CONNECTION TO THE AES ///////////////////////////////////////////////////////////////
 673 ////////////////////////////////////////////////////////////////////////////////////////////////////////
 674 ////////////////////////////////////////////////////////////////////////////////////////////////////////
 675
 676 void sendAESInformation (TServiceId sid)
 677 {
 678         AESIT aesit = findAES (sid);
 679
 680         vector<string> information;
 681
 682         CMessage msgout("AES_INFO");
 683
 684         //
 685         // send services that should be running on this AES
 686         //
 687         information.clear ();
 688         MYSQL_ROW row = sqlQuery ("select name from service where server='%s'", (*aesit).Name.c_str());
 689         while (row != NULL)
 690         {
 691                 string service = row[0];
 692                 nlinfo ("Adding '%s' in registered services to AES-%hu", row[0], sid.get());
 693                 information.push_back (service);
 694                 row = sqlNextRow ();
 695         }
 696         sqlFlushResult();
 697         msgout.serialCont (information);
 698
 699         //
 700         // send variable alarms for services that should running on this AES
 701         //
 702         information.clear ();
 703         row = sqlQuery ("select path, error_bound, alarm_order from variable where error_bound!=-1");
 704         while (row != NULL)
 705         {
 706                 nlinfo ("Adding '%s' '%s' '%s' in alarm to AES-%hu", row[0], row[1], row[2], sid.get());
 707                 information.push_back (row[0]);
 708                 information.push_back (row[1]);
 709                 information.push_back (row[2]);
 710                 row = sqlNextRow ();
 711         }
 712         sqlFlushResult();
 713         msgout.serialCont (information);
 714
 715         //
 716         // send graph update for services that should running on this AES
 717         //
 718         information.clear ();
 719         row = sqlQuery ("select path, graph_update from variable where graph_update!=0");
 720         while (row != NULL)
 721         {
 722                 CVarPath varpath (row[0]);
 723
 724                 for(uint i = 0; i < varpath.Destination.size(); i++)
 725                 {
 726                         string a  = varpath.Destination[i].first, b = (*aesit).Shard;
 727                         if(varpath.Destination[i].first == "*" || varpath.Destination[i].first == (*aesit).Shard)
 728                         {
 729                                 CVarPath varpath2 (varpath.Destination[i].second);
 730
 731                                 for(uint j = 0; j < varpath2.Destination.size(); j++)
 732                                 {
 733                                         string c  = varpath2.Destination[j].first, d = (*aesit).Name;
 734                                         if(varpath2.Destination[j].first == "*" || varpath2.Destination[j].first == (*aesit).Name)
 735                                         {
 736                                                 nlinfo ("Adding '%s' '%s' in graph to AES-%hu", row[0], row[1], sid.get());
 737                                                 information.push_back (row[0]);
 738                                                 information.push_back (row[1]);
 739                                         }
 740                                 }
 741                         }
 742                 }
 743                 row = sqlNextRow ();
 744         }
 745         sqlFlushResult();
 746         msgout.serialCont (information);
 747
 748         nlinfo ("Sending all information about %s AES-%hu (hostedservices, alarms,grapupdate)", (*aesit).Name.c_str(), (*aesit).SId.get());
 749         CUnifiedNetwork::getInstance ()->send (sid, msgout);
 750 }
 751
 752 void rejectAES(TServiceId sid, const string &res)
 753 {
 754         CMessage msgout("REJECTED");
 755         msgout.serial ((string &)res);
 756         CUnifiedNetwork::getInstance ()->send (sid, msgout);
 757 }
 758
 759 // i'm connected to a new admin executor service
 760 static void cbNewAESConnection (const std::string &serviceName, TServiceId sid, void *arg)
 761 {
 762         TSockId from;
 763         CCallbackNetBase *cnb = CUnifiedNetwork::getInstance ()->getNetBase (sid, from);
 764         const CInetAddress &ia = cnb->hostAddress (from);
 765
 766         AESIT aesit = findAES (sid, false);
 767
 768         if (aesit != AdminExecutorServices.end ())
 769         {
 770                 nlwarning ("Connection of an AES that is already in the list (%s)", ia.asString ().c_str ());
 771                 rejectAES (sid, "This AES is already in the AS list");
 772                 return;
 773         }
 774
 775         MYSQL_ROW row = sqlQuery ("select name from server where address='%s'", ia.ipAddress().c_str());
 776         if (row == NULL)
 777         {
 778                 if (!AllowExplicitAESRegistration)
 779                 {
 780                         nlwarning ("Connection of an AES that is not in database server list (%s)", ia.asString ().c_str ());
 781                 }
 782                 else
 783                 {
 784                         nlinfo ("Rejecting auto-connection of an AES (%s) - this should provke explicitly reconnect", ia.asString ().c_str ());
 785                 }
 786                 rejectAES (sid, "This AES is not registered in the database");
 787                 sqlFlushResult();
 788                 return;
 789         }
 790         string server = row[0];
 791         sqlFlushResult();
 792
 793         row = sqlQuery ("select shard from service where server='%s'", server.c_str());
 794         if (row == NULL)
 795         {
 796                 nlwarning ("Connection of an AES that is not in database server list (%s)", ia.asString ().c_str ());
 797                 rejectAES (sid, "This AES is not registered in the database");
 798                 sqlFlushResult();
 799                 return;
 800         }
 801         string shard = row[0];
 802         sqlFlushResult();
 803
 804         AdminExecutorServices.push_back (CAdminExecutorService(shard, server, sid));
 805
 806         nlinfo ("%s-%hu, server name %s, for shard %s connected and added in the list", serviceName.c_str(), sid.get(), server.c_str(), shard.c_str());
 807
 808         // send him services that should run on this server
 809         sendAESInformation (sid);
 810 }
 811
 812 // i'm disconnected from an admin executor service
 813 static void cbNewAESDisconnection (const std::string &serviceName, TServiceId sid, void *arg)
 814 {
 815         TSockId from;
 816         CCallbackNetBase *cnb = CUnifiedNetwork::getInstance ()->getNetBase (sid, from);
 817         const CInetAddress &ia = cnb->hostAddress (from);
 818
 819         AESIT aesit = findAES (sid, false);
 820
 821         if (aesit == AdminExecutorServices.end ())
 822         {
 823                 nlwarning ("Disconnection of %s-%hu that is not in my list (%s)", serviceName.c_str (), sid.get(), ia.asString ().c_str ());
 824                 return;
 825         }
 826
 827         nlinfo ("%s-%hu, shard name %s, disconnected and removed from the list", serviceName.c_str(), sid.get(), (*aesit).Name.c_str ());
 828
 829         // we need to remove pending request
 830
 831         for(uint i = 0; i < (*aesit).WaitingRequestId.size (); i++)
 832         {
 833                 subRequestWaitingNb ((*aesit).WaitingRequestId[i]);
 834         }
 835
 836         AdminExecutorServices.erase (aesit);
 837 }
 838
 839 // we receive an explicit registration message from an AES
 840 void cbRegisterAES(CMessage &msgin, const std::string &serviceName, TServiceId sid)
 841 {
 842         if (!AllowExplicitAESRegistration)
 843         {
 844                 nlwarning("Ignoring attempted AES registration because AllowExplicitAESRegistration==false");
 845                 return;
 846         }
 847
 848         string server;
 849         string shard;
 850         try
 851         {
 852                 msgin.serial(server);
 853                 msgin.serial(shard);
 854         }
 855         catch(...)
 856         {
 857                 nlwarning("Ignoring attempted AES registration due to execption during message decoding");
 858                 return;
 859         }
 860
 861         AdminExecutorServices.push_back (CAdminExecutorService(shard, server, sid));
 862
 863         nlinfo ("%s-%hu, server name %s, for shard %s connected and added in the list", serviceName.c_str(), sid.get(), server.c_str(), shard.c_str());
 864
 865         // send him services that should run on this server
 866         sendAESInformation (sid);
 867 }
 868
 869 static void cbView (CMessage &msgin, const std::string &serviceName, TServiceId sid)
 870 {
 871         uint32 rid;
 872         msgin.serial (rid);
 873
 874         AESIT aesit = findAES (sid);
 875
 876         for (uint i = 0; i < (*aesit).WaitingRequestId.size();)
 877         {
 878                 if ((*aesit).WaitingRequestId[i] == rid)
 879                 {
 880                         (*aesit).WaitingRequestId.erase ((*aesit).WaitingRequestId.begin ()+i);
 881                 }
 882                 else
 883                 {
 884                         i++;
 885                 }
 886         }
 887
 888         MYSQL_ROW row = sqlQuery ("select distinct shard from service where server='%s'", (*aesit).Name.c_str ());
 889
 890         // shard name is find using the "service" table, so, if there s no shard name in it, it returns ???
 891         string shardName;
 892         if (row != NULL) shardName = row[0];
 893         else shardName = DontUseDataBase? aesit->Shard: "???";
 894
 895         vector<string> vara, vala;
 896
 897         while ((uint32)msgin.getPos() < msgin.length())
 898         {
 899                 vara.clear ();
 900                 vala.clear ();
 901
 902                 // adding default row
 903                 vara.push_back ("shard");
 904                 vara.push_back ("server");
 905
 906                 vala.push_back (shardName);
 907                 vala.push_back ((*aesit).Name);
 908
 909                 uint32 i, nb;
 910                 string var, val;
 911
 912                 msgin.serial (nb);
 913                 for (i = 0; i < nb; i++)
 914                 {
 915                         msgin.serial (var);
 916                         if (var == "__log")
 917                         {
 918                                 vara.clear ();
 919                                 vala.clear ();
 920                         }
 921                         vara.push_back (var);
 922                 }
 923
 924                 if (vara.size() > 0 && vara[0] == "__log")
 925                         vala.push_back ("----- Result from Shard "+shardName+" Server "+(*aesit).Name+"\n");
 926
 927                 msgin.serial (nb);
 928                 for (i = 0; i < nb; i++)
 929                 {
 930                         msgin.serial (val);
 931                         vala.push_back (val);
 932                 }
 933                 addRequestAnswer (rid, vara, vala);
 934         }
 935         sqlFlushResult();
 936
 937         // inc the NbReceived counter
 938         addRequestReceived (rid);
 939 }
 940
 941 TUnifiedCallbackItem CallbackArray[] =
 942 {
 943         { "REGISTER_AES", cbRegisterAES },
 944         { "VIEW", cbView },
 945         { "ADMIN_EMAIL", cbAdminEmail },
 946         { "GRAPH_UPDATE", cbGraphUpdate },
 947 };
 948
 949
 950 ////////////////////////////////////////////////////////////////////////////////////////////////////////
 951 ////////////////////////////////////////////////////////////////////////////////////////////////////////
 952 ////////////////// CONNECTION TO THE CLIENT ////////////////////////////////////////////////////////////
 953 ////////////////////////////////////////////////////////////////////////////////////////////////////////
 954 ////////////////////////////////////////////////////////////////////////////////////////////////////////
 955
 956 void addRequest (const string &rawvarpath, TSockId from)
 957 {
 958         nlinfo ("addRequest from %s: '%s'", from->asString ().c_str (), rawvarpath.c_str ());
 959
 960         if(rawvarpath.empty ())
 961         {
 962                 // send an empty string to say to php that there's nothing
 963                 string str;
 964                 sendString (from, str);
 965         }
 966
 967         //
 968         // special cases
 969         //
 970
 971         if(rawvarpath == "reload")
 972         {
 973                 // it means the we have to resend the list of services managed by AES from the mysql tables
 974                 for (AESIT aesit = AdminExecutorServices.begin(); aesit != AdminExecutorServices.end(); aesit++)
 975                 {
 976                         sendAESInformation ((*aesit).SId);
 977                 }
 978
 979                 // send an empty string to say to php that there's nothing
 980                 string str;
 981                 sendString (from, str);
 982                 return;
 983         }
 984
 985         //
 986         // normal cases
 987         //
 988
 989         CVarPath varpath (rawvarpath);
 990
 991         uint32 rid = newRequest (from);
 992
 993         for (uint i = 0; i < varpath.Destination.size (); i++)
 994         {
 995                 string shard = varpath.Destination[i].first;
 996
 997                 CVarPath subvarpath (varpath.Destination[i].second);
 998
 999                 for (uint j = 0; j < subvarpath.Destination.size (); j++)
1000                 {
1001                         string server = subvarpath.Destination[j].first;
1002
1003                         if (shard == "*" && server == "*")
1004                         {
1005                                 // Send the request to all online servers of all online shards
1006
1007                                 AESIT aesit;
1008                                 for (aesit = AdminExecutorServices.begin(); aesit != AdminExecutorServices.end(); aesit++)
1009                                 {
1010                                         addRequestWaitingNb (rid);
1011                                         (*aesit).WaitingRequestId.push_back (rid);
1012
1013                                         CMessage msgout("AES_GET_VIEW");
1014                                         msgout.serial (rid);
1015                                         msgout.serial (subvarpath.Destination[j].second);
1016                                         CUnifiedNetwork::getInstance ()->send ((*aesit).SId, msgout);
1017                                         nlinfo ("REQUEST: Sent view '%s' to shard name %s 'AES-%hu'", subvarpath.Destination[j].second.c_str(), (*aesit).Name.c_str(), (*aesit).SId.get());
1018                                 }
1019                         }
1020                         else if (shard == "*" && server == "#")
1021                         {
1022                                 // Select all shard all server including offline one
1023
1024                                 MYSQL_ROW row = sqlQuery ("select distinct server, shard from service");
1025
1026                                 while (row != NULL)
1027                                 {
1028                                         AESIT aesit = findAES (row[0], false);
1029
1030                                         if (aesit != AdminExecutorServices.end())
1031                                         {
1032                                                 addRequestWaitingNb (rid);
1033                                                 (*aesit).WaitingRequestId.push_back (rid);
1034
1035                                                 CMessage msgout("AES_GET_VIEW");
1036                                                 msgout.serial (rid);
1037                                                 msgout.serial (subvarpath.Destination[j].second);
1038                                                 CUnifiedNetwork::getInstance ()->send ((*aesit).SId, msgout);
1039                                                 nlinfo ("REQUEST: Sent view '%s' to shard name %s 'AES-%hu'", subvarpath.Destination[j].second.c_str(), (*aesit).Name.c_str(), (*aesit).SId.get());
1040
1041                                         }
1042                                         else if (server == "#")
1043                                         {
1044                                                 vector<string> vara, vala;
1045
1046                                                 // adding default row
1047                                                 vara.push_back ("shard");
1048                                                 vala.push_back (row[1]);
1049
1050                                                 vara.push_back ("server");
1051                                                 vala.push_back (row[0]);
1052
1053                                                 vara.push_back ("service");
1054                                                 vala.push_back ("AES");
1055
1056                                                 vara.push_back ("State");
1057                                                 vala.push_back ("Offline");
1058
1059                                                 addRequestAnswer (rid, vara, vala);
1060                                         }
1061                                         row = sqlNextRow ();
1062                                 }
1063                                 sqlFlushResult();
1064                         }
1065                         else if (server == "*" || server == "#")
1066                         {
1067                                 // Send the request to all online server of a specific shard
1068
1069                                 MYSQL_ROW row = sqlQuery ("select distinct server from service where shard='%s'", shard.c_str ());
1070
1071                                 while (row != NULL)
1072                                 {
1073                                         AESIT aesit = findAES (row[0], false);
1074
1075                                         if (aesit != AdminExecutorServices.end())
1076                                         {
1077                                                 addRequestWaitingNb (rid);
1078                                                 (*aesit).WaitingRequestId.push_back (rid);
1079
1080                                                 CMessage msgout("AES_GET_VIEW");
1081                                                 msgout.serial (rid);
1082                                                 msgout.serial (subvarpath.Destination[j].second);
1083                                                 CUnifiedNetwork::getInstance ()->send ((*aesit).SId, msgout);
1084                                                 nlinfo ("REQUEST: Sent view '%s' to shard name %s 'AES-%hu'", subvarpath.Destination[j].second.c_str(), (*aesit).Name.c_str(), (*aesit).SId.get());
1085
1086                                         }
1087                                         else if (server == "#")
1088                                         {
1089                                                 vector<string> vara, vala;
1090
1091                                                 // adding default row
1092                                                 vara.push_back ("shard");
1093                                                 vala.push_back (shard);
1094
1095                                                 vara.push_back ("server");
1096                                                 vala.push_back (row[0]);
1097
1098                                                 vara.push_back ("service");
1099                                                 vala.push_back ("AES");
1100
1101                                                 vara.push_back ("State");
1102                                                 vala.push_back ("Offline");
1103
1104                                                 addRequestAnswer (rid, vara, vala);
1105                                         }
1106                                         row = sqlNextRow ();
1107                                 }
1108
1109                                 sqlFlushResult();
1110                         }
1111                         else
1112                         {
1113                                 AESIT aesit = findAES (server, false);
1114
1115                                 if (aesit != AdminExecutorServices.end())
1116                                 {
1117                                         addRequestWaitingNb (rid);
1118                                         (*aesit).WaitingRequestId.push_back (rid);
1119
1120                                         CMessage msgout("AES_GET_VIEW");
1121                                         msgout.serial (rid);
1122                                         msgout.serial (subvarpath.Destination[j].second);
1123                                         CUnifiedNetwork::getInstance ()->send ((*aesit).SId, msgout);
1124                                         nlinfo ("REQUEST: Sent view '%s' to shard name %s 'AES-%hu'", subvarpath.Destination[j].second.c_str(), (*aesit).Name.c_str(), (*aesit).SId.get());
1125                                 }
1126                                 else
1127                                 {
1128                                         nlwarning ("Server %s is not found in the list", server.c_str ());
1129                                 }
1130                         }
1131                 }
1132         }
1133 }
1134
1135 static void varRequestTimeout(CConfigFile::CVar &var)
1136 {
1137         RequestTimeout = var.asInt();
1138         nlinfo ("Request timeout is now after %d seconds", RequestTimeout);
1139 }
1140
1141 static void varAdminAlertAccumlationTime (CConfigFile::CVar &var)
1142 {
1143         AdminAlertAccumlationTime = var.asInt();
1144 }
1145
1146
1147 ////////////////////////////////////////////////////////////////////////////////////////////////////////
1148 ////////////////////////////////////////////////////////////////////////////////////////////////////////
1149 ////////////////// SERVICE IMPLEMENTATION //////////////////////////////////////////////////////////////
1150 ////////////////////////////////////////////////////////////////////////////////////////////////////////
1151 ////////////////////////////////////////////////////////////////////////////////////////////////////////
1152
1153 class CAdminService : public IService
1154 {
1155 public:
1156
1157         /// Init the service, load the universal time.
1158         void init ()
1159         {
1160                 setDefaultEmailParams (ConfigFile.getVar ("SMTPServer").asString (), ConfigFile.getVar("DefaultEmailFrom").asString(), "");
1161
1162                 sqlInit ();
1163
1164                 connectionWebInit ();
1165
1166                 //CVarPath toto ("[toto");
1167
1168                 //CVarPath toto ("*.*.*.*");
1169                 //CVarPath toto ("[srv1,srv2].*.*.*");
1170                 //CVarPath toto ("[svr1.svc1,srv2.svc2].*.*");
1171                 //CVarPath toto ("[svr1.[svc1,svc2].*.var1,srv2.svc2.fe*.var2].toto");
1172                 //CVarPath toto ("[svr1.svc1.*.toto,srv2.svc2.*.tata]");
1173
1174                 CUnifiedNetwork::getInstance ()->setServiceUpCallback ("AES", cbNewAESConnection);
1175                 CUnifiedNetwork::getInstance ()->setServiceDownCallback ("AES", cbNewAESDisconnection);
1176
1177                 varRequestTimeout (ConfigFile.getVar ("RequestTimeout"));
1178                 ConfigFile.setCallback("RequestTimeout", &varRequestTimeout);
1179
1180                 varAdminAlertAccumlationTime (ConfigFile.getVar ("AdminAlertAccumlationTime"));
1181                 ConfigFile.setCallback("AdmimAlertAccumlationTime", &varAdminAlertAccumlationTime);
1182
1183         }
1184
1185         bool update ()
1186         {
1187                 cleanRequest ();
1188                 connectionWebUpdate ();
1189
1190                 updateSendAdminAlert ();
1191                 return true;
1192         }
1193
1194         void release ()
1195         {
1196                 connectionWebRelease ();
1197         }
1198 };
1199
1200
1201 /// Admin Service
1202 NLNET_SERVICE_MAIN (CAdminService, "AS", "admin_service", 49996, CallbackArray, NELNS_CONFIG, NELNS_LOGS);
1203
1204
1205 NLMISC_COMMAND (getViewAS, "send a view and receive an array as result", "<varpath>")
1206 {
1207         string cmd;
1208         for (uint i = 0; i < args.size(); i++)
1209         {
1210                 if (i != 0) cmd += " ";
1211                 cmd += args[i];
1212         }
1213
1214         addRequest (cmd, NULL);
1215
1216         return true;
1217 }
1218
1219 NLMISC_COMMAND (clearRequests, "clear all pending requests", "")
1220 {
1221         if(args.size() != 0) return false;
1222
1223         // for all request, set the NbWaiting to NbReceived, next cleanRequest() will send answer and clear all request
1224         for (uint i = 0 ; i < Requests.size (); i++)
1225         {
1226                 if (Requests[i].NbWaiting <= Requests[i].NbReceived)
1227                 {
1228                         Requests[i].NbWaiting = Requests[i].NbReceived;
1229                 }
1230         }
1231
1232         return true;
1233 }
1234
1235 NLMISC_COMMAND (displayRequests, "display all pending requests", "")
1236 {
1237         if(args.size() != 0) return false;
1238
1239         log.displayNL ("Display %d pending requests", Requests.size ());
1240         for (uint i = 0 ; i < Requests.size (); i++)
1241         {
1242                 log.displayNL ("id: %d wait: %d recv: %d from: %s nbrow: %d", Requests[i].Id, Requests[i].NbWaiting, Requests[i].NbReceived, Requests[i].From->asString ().c_str (), Requests[i].NbRow);
1243         }
1244         log.displayNL ("End of display pending requests");
1245
1246         return true;
1247 }
1248
1249 NLMISC_COMMAND (generateAlert, "generate an alert", "<text>")
1250 {
1251         if(args.size() != 1) return false;
1252
1253         sendAdminAlert (args[0].c_str());
1254
1255         return true;
1256 }