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 #if LIBMYSQL_VERSION_ID < 80000
 633         my_bool opt = true;
 634 #else
 635         bool opt = true;
 636 #endif
 637         if (mysql_options (db, MYSQL_OPT_RECONNECT, &opt))
 638         {
 639                 mysql_close(db);
 640                 DatabaseConnection = 0;
 641                 nlerror("mysql_options() failed for database connection to '%s'", IService::getInstance()->ConfigFile.getVar("DatabaseHost").asString().c_str());
 642                 return;
 643         }
 644
 645         DatabaseConnection = mysql_real_connect(db,
 646                 IService::getInstance()->ConfigFile.getVar("DatabaseHost").asString().c_str(),
 647                 IService::getInstance()->ConfigFile.getVar("DatabaseLogin").asString().c_str(),
 648                 IService::getInstance()->ConfigFile.getVar("DatabasePassword").asString().c_str(),
 649                 IService::getInstance()->ConfigFile.getVar("DatabaseName").asString().c_str(),
 650                 0,NULL,0);
 651         if (DatabaseConnection == NULL || DatabaseConnection != db)
 652         {
 653                 nlerror ("mysql_real_connect() failed to '%s' with login '%s' and database name '%s' with %s",
 654                         IService::getInstance()->ConfigFile.getVar("DatabaseHost").asString().c_str(),
 655                         IService::getInstance()->ConfigFile.getVar("DatabaseLogin").asString().c_str(),
 656                         IService::getInstance()->ConfigFile.getVar("DatabaseName").asString().c_str(),
 657                         (IService::getInstance()->ConfigFile.getVar("DatabasePassword").asString().empty()?"empty password":"password")
 658                         );
 659         }
 660
 661 #if MYSQL_VERSION_ID < 50019
 662         opt = true;
 663         if (mysql_options (DatabaseConnection, MYSQL_OPT_RECONNECT, &opt))
 664         {
 665                 mysql_close(db);
 666                 DatabaseConnection = 0;
 667                 nlerror("mysql_options() failed for database connection to '%s'", IService::getInstance()->ConfigFile.getVar("DatabaseHost").asString().c_str());
 668                 return;
 669         }
 670 #endif
 671 }
 672
 673
 674 ////////////////////////////////////////////////////////////////////////////////////////////////////////
 675 ////////////////////////////////////////////////////////////////////////////////////////////////////////
 676 ////////////////// CONNECTION TO THE AES ///////////////////////////////////////////////////////////////
 677 ////////////////////////////////////////////////////////////////////////////////////////////////////////
 678 ////////////////////////////////////////////////////////////////////////////////////////////////////////
 679
 680 void sendAESInformation (TServiceId sid)
 681 {
 682         AESIT aesit = findAES (sid);
 683
 684         vector<string> information;
 685
 686         CMessage msgout("AES_INFO");
 687
 688         //
 689         // send services that should be running on this AES
 690         //
 691         information.clear ();
 692         MYSQL_ROW row = sqlQuery ("select name from service where server='%s'", (*aesit).Name.c_str());
 693         while (row != NULL)
 694         {
 695                 string service = row[0];
 696                 nlinfo ("Adding '%s' in registered services to AES-%hu", row[0], sid.get());
 697                 information.push_back (service);
 698                 row = sqlNextRow ();
 699         }
 700         sqlFlushResult();
 701         msgout.serialCont (information);
 702
 703         //
 704         // send variable alarms for services that should running on this AES
 705         //
 706         information.clear ();
 707         row = sqlQuery ("select path, error_bound, alarm_order from variable where error_bound!=-1");
 708         while (row != NULL)
 709         {
 710                 nlinfo ("Adding '%s' '%s' '%s' in alarm to AES-%hu", row[0], row[1], row[2], sid.get());
 711                 information.push_back (row[0]);
 712                 information.push_back (row[1]);
 713                 information.push_back (row[2]);
 714                 row = sqlNextRow ();
 715         }
 716         sqlFlushResult();
 717         msgout.serialCont (information);
 718
 719         //
 720         // send graph update for services that should running on this AES
 721         //
 722         information.clear ();
 723         row = sqlQuery ("select path, graph_update from variable where graph_update!=0");
 724         while (row != NULL)
 725         {
 726                 CVarPath varpath (row[0]);
 727
 728                 for(uint i = 0; i < varpath.Destination.size(); i++)
 729                 {
 730                         string a  = varpath.Destination[i].first, b = (*aesit).Shard;
 731                         if(varpath.Destination[i].first == "*" || varpath.Destination[i].first == (*aesit).Shard)
 732                         {
 733                                 CVarPath varpath2 (varpath.Destination[i].second);
 734
 735                                 for(uint j = 0; j < varpath2.Destination.size(); j++)
 736                                 {
 737                                         string c  = varpath2.Destination[j].first, d = (*aesit).Name;
 738                                         if(varpath2.Destination[j].first == "*" || varpath2.Destination[j].first == (*aesit).Name)
 739                                         {
 740                                                 nlinfo ("Adding '%s' '%s' in graph to AES-%hu", row[0], row[1], sid.get());
 741                                                 information.push_back (row[0]);
 742                                                 information.push_back (row[1]);
 743                                         }
 744                                 }
 745                         }
 746                 }
 747                 row = sqlNextRow ();
 748         }
 749         sqlFlushResult();
 750         msgout.serialCont (information);
 751
 752         nlinfo ("Sending all information about %s AES-%hu (hostedservices, alarms,grapupdate)", (*aesit).Name.c_str(), (*aesit).SId.get());
 753         CUnifiedNetwork::getInstance ()->send (sid, msgout);
 754 }
 755
 756 void rejectAES(TServiceId sid, const string &res)
 757 {
 758         CMessage msgout("REJECTED");
 759         msgout.serial ((string &)res);
 760         CUnifiedNetwork::getInstance ()->send (sid, msgout);
 761 }
 762
 763 // i'm connected to a new admin executor service
 764 static void cbNewAESConnection (const std::string &serviceName, TServiceId sid, void *arg)
 765 {
 766         TSockId from;
 767         CCallbackNetBase *cnb = CUnifiedNetwork::getInstance ()->getNetBase (sid, from);
 768         const CInetAddress &ia = cnb->hostAddress (from);
 769
 770         AESIT aesit = findAES (sid, false);
 771
 772         if (aesit != AdminExecutorServices.end ())
 773         {
 774                 nlwarning ("Connection of an AES that is already in the list (%s)", ia.asString ().c_str ());
 775                 rejectAES (sid, "This AES is already in the AS list");
 776                 return;
 777         }
 778
 779         MYSQL_ROW row = sqlQuery ("select name from server where address='%s'", ia.ipAddress().c_str());
 780         if (row == NULL)
 781         {
 782                 if (!AllowExplicitAESRegistration)
 783                 {
 784                         nlwarning ("Connection of an AES that is not in database server list (%s)", ia.asString ().c_str ());
 785                 }
 786                 else
 787                 {
 788                         nlinfo ("Rejecting auto-connection of an AES (%s) - this should provke explicitly reconnect", ia.asString ().c_str ());
 789                 }
 790                 rejectAES (sid, "This AES is not registered in the database");
 791                 sqlFlushResult();
 792                 return;
 793         }
 794         string server = row[0];
 795         sqlFlushResult();
 796
 797         row = sqlQuery ("select shard from service where server='%s'", server.c_str());
 798         if (row == NULL)
 799         {
 800                 nlwarning ("Connection of an AES that is not in database server list (%s)", ia.asString ().c_str ());
 801                 rejectAES (sid, "This AES is not registered in the database");
 802                 sqlFlushResult();
 803                 return;
 804         }
 805         string shard = row[0];
 806         sqlFlushResult();
 807
 808         AdminExecutorServices.push_back (CAdminExecutorService(shard, server, sid));
 809
 810         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());
 811
 812         // send him services that should run on this server
 813         sendAESInformation (sid);
 814 }
 815
 816 // i'm disconnected from an admin executor service
 817 static void cbNewAESDisconnection (const std::string &serviceName, TServiceId sid, void *arg)
 818 {
 819         TSockId from;
 820         CCallbackNetBase *cnb = CUnifiedNetwork::getInstance ()->getNetBase (sid, from);
 821         const CInetAddress &ia = cnb->hostAddress (from);
 822
 823         AESIT aesit = findAES (sid, false);
 824
 825         if (aesit == AdminExecutorServices.end ())
 826         {
 827                 nlwarning ("Disconnection of %s-%hu that is not in my list (%s)", serviceName.c_str (), sid.get(), ia.asString ().c_str ());
 828                 return;
 829         }
 830
 831         nlinfo ("%s-%hu, shard name %s, disconnected and removed from the list", serviceName.c_str(), sid.get(), (*aesit).Name.c_str ());
 832
 833         // we need to remove pending request
 834
 835         for(uint i = 0; i < (*aesit).WaitingRequestId.size (); i++)
 836         {
 837                 subRequestWaitingNb ((*aesit).WaitingRequestId[i]);
 838         }
 839
 840         AdminExecutorServices.erase (aesit);
 841 }
 842
 843 // we receive an explicit registration message from an AES
 844 void cbRegisterAES(CMessage &msgin, const std::string &serviceName, TServiceId sid)
 845 {
 846         if (!AllowExplicitAESRegistration)
 847         {
 848                 nlwarning("Ignoring attempted AES registration because AllowExplicitAESRegistration==false");
 849                 return;
 850         }
 851
 852         string server;
 853         string shard;
 854         try
 855         {
 856                 msgin.serial(server);
 857                 msgin.serial(shard);
 858         }
 859         catch(...)
 860         {
 861                 nlwarning("Ignoring attempted AES registration due to execption during message decoding");
 862                 return;
 863         }
 864
 865         AdminExecutorServices.push_back (CAdminExecutorService(shard, server, sid));
 866
 867         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());
 868
 869         // send him services that should run on this server
 870         sendAESInformation (sid);
 871 }
 872
 873 static void cbView (CMessage &msgin, const std::string &serviceName, TServiceId sid)
 874 {
 875         uint32 rid;
 876         msgin.serial (rid);
 877
 878         AESIT aesit = findAES (sid);
 879
 880         for (uint i = 0; i < (*aesit).WaitingRequestId.size();)
 881         {
 882                 if ((*aesit).WaitingRequestId[i] == rid)
 883                 {
 884                         (*aesit).WaitingRequestId.erase ((*aesit).WaitingRequestId.begin ()+i);
 885                 }
 886                 else
 887                 {
 888                         i++;
 889                 }
 890         }
 891
 892         MYSQL_ROW row = sqlQuery ("select distinct shard from service where server='%s'", (*aesit).Name.c_str ());
 893
 894         // shard name is find using the "service" table, so, if there s no shard name in it, it returns ???
 895         string shardName;
 896         if (row != NULL) shardName = row[0];
 897         else shardName = DontUseDataBase? aesit->Shard: "???";
 898
 899         vector<string> vara, vala;
 900
 901         while ((uint32)msgin.getPos() < msgin.length())
 902         {
 903                 vara.clear ();
 904                 vala.clear ();
 905
 906                 // adding default row
 907                 vara.push_back ("shard");
 908                 vara.push_back ("server");
 909
 910                 vala.push_back (shardName);
 911                 vala.push_back ((*aesit).Name);
 912
 913                 uint32 i, nb;
 914                 string var, val;
 915
 916                 msgin.serial (nb);
 917                 for (i = 0; i < nb; i++)
 918                 {
 919                         msgin.serial (var);
 920                         if (var == "__log")
 921                         {
 922                                 vara.clear ();
 923                                 vala.clear ();
 924                         }
 925                         vara.push_back (var);
 926                 }
 927
 928                 if (vara.size() > 0 && vara[0] == "__log")
 929                         vala.push_back ("----- Result from Shard "+shardName+" Server "+(*aesit).Name+"\n");
 930
 931                 msgin.serial (nb);
 932                 for (i = 0; i < nb; i++)
 933                 {
 934                         msgin.serial (val);
 935                         vala.push_back (val);
 936                 }
 937                 addRequestAnswer (rid, vara, vala);
 938         }
 939         sqlFlushResult();
 940
 941         // inc the NbReceived counter
 942         addRequestReceived (rid);
 943 }
 944
 945 TUnifiedCallbackItem CallbackArray[] =
 946 {
 947         { "REGISTER_AES", cbRegisterAES },
 948         { "VIEW", cbView },
 949         { "ADMIN_EMAIL", cbAdminEmail },
 950         { "GRAPH_UPDATE", cbGraphUpdate },
 951 };
 952
 953
 954 ////////////////////////////////////////////////////////////////////////////////////////////////////////
 955 ////////////////////////////////////////////////////////////////////////////////////////////////////////
 956 ////////////////// CONNECTION TO THE CLIENT ////////////////////////////////////////////////////////////
 957 ////////////////////////////////////////////////////////////////////////////////////////////////////////
 958 ////////////////////////////////////////////////////////////////////////////////////////////////////////
 959
 960 void addRequest (const string &rawvarpath, TSockId from)
 961 {
 962         nlinfo ("addRequest from %s: '%s'", from->asString ().c_str (), rawvarpath.c_str ());
 963
 964         if(rawvarpath.empty ())
 965         {
 966                 // send an empty string to say to php that there's nothing
 967                 string str;
 968                 sendString (from, str);
 969         }
 970
 971         //
 972         // special cases
 973         //
 974
 975         if(rawvarpath == "reload")
 976         {
 977                 // it means the we have to resend the list of services managed by AES from the mysql tables
 978                 for (AESIT aesit = AdminExecutorServices.begin(); aesit != AdminExecutorServices.end(); aesit++)
 979                 {
 980                         sendAESInformation ((*aesit).SId);
 981                 }
 982
 983                 // send an empty string to say to php that there's nothing
 984                 string str;
 985                 sendString (from, str);
 986                 return;
 987         }
 988
 989         //
 990         // normal cases
 991         //
 992
 993         CVarPath varpath (rawvarpath);
 994
 995         uint32 rid = newRequest (from);
 996
 997         for (uint i = 0; i < varpath.Destination.size (); i++)
 998         {
 999                 string shard = varpath.Destination[i].first;
1000
1001                 CVarPath subvarpath (varpath.Destination[i].second);
1002
1003                 for (uint j = 0; j < subvarpath.Destination.size (); j++)
1004                 {
1005                         string server = subvarpath.Destination[j].first;
1006
1007                         if (shard == "*" && server == "*")
1008                         {
1009                                 // Send the request to all online servers of all online shards
1010
1011                                 AESIT aesit;
1012                                 for (aesit = AdminExecutorServices.begin(); aesit != AdminExecutorServices.end(); aesit++)
1013                                 {
1014                                         addRequestWaitingNb (rid);
1015                                         (*aesit).WaitingRequestId.push_back (rid);
1016
1017                                         CMessage msgout("AES_GET_VIEW");
1018                                         msgout.serial (rid);
1019                                         msgout.serial (subvarpath.Destination[j].second);
1020                                         CUnifiedNetwork::getInstance ()->send ((*aesit).SId, msgout);
1021                                         nlinfo ("REQUEST: Sent view '%s' to shard name %s 'AES-%hu'", subvarpath.Destination[j].second.c_str(), (*aesit).Name.c_str(), (*aesit).SId.get());
1022                                 }
1023                         }
1024                         else if (shard == "*" && server == "#")
1025                         {
1026                                 // Select all shard all server including offline one
1027
1028                                 MYSQL_ROW row = sqlQuery ("select distinct server, shard from service");
1029
1030                                 while (row != NULL)
1031                                 {
1032                                         AESIT aesit = findAES (row[0], false);
1033
1034                                         if (aesit != AdminExecutorServices.end())
1035                                         {
1036                                                 addRequestWaitingNb (rid);
1037                                                 (*aesit).WaitingRequestId.push_back (rid);
1038
1039                                                 CMessage msgout("AES_GET_VIEW");
1040                                                 msgout.serial (rid);
1041                                                 msgout.serial (subvarpath.Destination[j].second);
1042                                                 CUnifiedNetwork::getInstance ()->send ((*aesit).SId, msgout);
1043                                                 nlinfo ("REQUEST: Sent view '%s' to shard name %s 'AES-%hu'", subvarpath.Destination[j].second.c_str(), (*aesit).Name.c_str(), (*aesit).SId.get());
1044
1045                                         }
1046                                         else if (server == "#")
1047                                         {
1048                                                 vector<string> vara, vala;
1049
1050                                                 // adding default row
1051                                                 vara.push_back ("shard");
1052                                                 vala.push_back (row[1]);
1053
1054                                                 vara.push_back ("server");
1055                                                 vala.push_back (row[0]);
1056
1057                                                 vara.push_back ("service");
1058                                                 vala.push_back ("AES");
1059
1060                                                 vara.push_back ("State");
1061                                                 vala.push_back ("Offline");
1062
1063                                                 addRequestAnswer (rid, vara, vala);
1064                                         }
1065                                         row = sqlNextRow ();
1066                                 }
1067                                 sqlFlushResult();
1068                         }
1069                         else if (server == "*" || server == "#")
1070                         {
1071                                 // Send the request to all online server of a specific shard
1072
1073                                 MYSQL_ROW row = sqlQuery ("select distinct server from service where shard='%s'", shard.c_str ());
1074
1075                                 while (row != NULL)
1076                                 {
1077                                         AESIT aesit = findAES (row[0], false);
1078
1079                                         if (aesit != AdminExecutorServices.end())
1080                                         {
1081                                                 addRequestWaitingNb (rid);
1082                                                 (*aesit).WaitingRequestId.push_back (rid);
1083
1084                                                 CMessage msgout("AES_GET_VIEW");
1085                                                 msgout.serial (rid);
1086                                                 msgout.serial (subvarpath.Destination[j].second);
1087                                                 CUnifiedNetwork::getInstance ()->send ((*aesit).SId, msgout);
1088                                                 nlinfo ("REQUEST: Sent view '%s' to shard name %s 'AES-%hu'", subvarpath.Destination[j].second.c_str(), (*aesit).Name.c_str(), (*aesit).SId.get());
1089
1090                                         }
1091                                         else if (server == "#")
1092                                         {
1093                                                 vector<string> vara, vala;
1094
1095                                                 // adding default row
1096                                                 vara.push_back ("shard");
1097                                                 vala.push_back (shard);
1098
1099                                                 vara.push_back ("server");
1100                                                 vala.push_back (row[0]);
1101
1102                                                 vara.push_back ("service");
1103                                                 vala.push_back ("AES");
1104
1105                                                 vara.push_back ("State");
1106                                                 vala.push_back ("Offline");
1107
1108                                                 addRequestAnswer (rid, vara, vala);
1109                                         }
1110                                         row = sqlNextRow ();
1111                                 }
1112
1113                                 sqlFlushResult();
1114                         }
1115                         else
1116                         {
1117                                 AESIT aesit = findAES (server, false);
1118
1119                                 if (aesit != AdminExecutorServices.end())
1120                                 {
1121                                         addRequestWaitingNb (rid);
1122                                         (*aesit).WaitingRequestId.push_back (rid);
1123
1124                                         CMessage msgout("AES_GET_VIEW");
1125                                         msgout.serial (rid);
1126                                         msgout.serial (subvarpath.Destination[j].second);
1127                                         CUnifiedNetwork::getInstance ()->send ((*aesit).SId, msgout);
1128                                         nlinfo ("REQUEST: Sent view '%s' to shard name %s 'AES-%hu'", subvarpath.Destination[j].second.c_str(), (*aesit).Name.c_str(), (*aesit).SId.get());
1129                                 }
1130                                 else
1131                                 {
1132                                         nlwarning ("Server %s is not found in the list", server.c_str ());
1133                                 }
1134                         }
1135                 }
1136         }
1137 }
1138
1139 static void varRequestTimeout(CConfigFile::CVar &var)
1140 {
1141         RequestTimeout = var.asInt();
1142         nlinfo ("Request timeout is now after %d seconds", RequestTimeout);
1143 }
1144
1145 static void varAdminAlertAccumlationTime (CConfigFile::CVar &var)
1146 {
1147         AdminAlertAccumlationTime = var.asInt();
1148 }
1149
1150
1151 ////////////////////////////////////////////////////////////////////////////////////////////////////////
1152 ////////////////////////////////////////////////////////////////////////////////////////////////////////
1153 ////////////////// SERVICE IMPLEMENTATION //////////////////////////////////////////////////////////////
1154 ////////////////////////////////////////////////////////////////////////////////////////////////////////
1155 ////////////////////////////////////////////////////////////////////////////////////////////////////////
1156
1157 class CAdminService : public IService
1158 {
1159 public:
1160
1161         /// Init the service, load the universal time.
1162         void init ()
1163         {
1164                 setDefaultEmailParams (ConfigFile.getVar ("SMTPServer").asString (), ConfigFile.getVar("DefaultEmailFrom").asString(), "");
1165
1166                 sqlInit ();
1167
1168                 connectionWebInit ();
1169
1170                 //CVarPath toto ("[toto");
1171
1172                 //CVarPath toto ("*.*.*.*");
1173                 //CVarPath toto ("[srv1,srv2].*.*.*");
1174                 //CVarPath toto ("[svr1.svc1,srv2.svc2].*.*");
1175                 //CVarPath toto ("[svr1.[svc1,svc2].*.var1,srv2.svc2.fe*.var2].toto");
1176                 //CVarPath toto ("[svr1.svc1.*.toto,srv2.svc2.*.tata]");
1177
1178                 CUnifiedNetwork::getInstance ()->setServiceUpCallback ("AES", cbNewAESConnection);
1179                 CUnifiedNetwork::getInstance ()->setServiceDownCallback ("AES", cbNewAESDisconnection);
1180
1181                 varRequestTimeout (ConfigFile.getVar ("RequestTimeout"));
1182                 ConfigFile.setCallback("RequestTimeout", &varRequestTimeout);
1183
1184                 varAdminAlertAccumlationTime (ConfigFile.getVar ("AdminAlertAccumlationTime"));
1185                 ConfigFile.setCallback("AdmimAlertAccumlationTime", &varAdminAlertAccumlationTime);
1186
1187         }
1188
1189         bool update ()
1190         {
1191                 cleanRequest ();
1192                 connectionWebUpdate ();
1193
1194                 updateSendAdminAlert ();
1195                 return true;
1196         }
1197
1198         void release ()
1199         {
1200                 connectionWebRelease ();
1201         }
1202 };
1203
1204
1205 /// Admin Service
1206 NLNET_SERVICE_MAIN (CAdminService, "AS", "admin_service", 49996, CallbackArray, NELNS_CONFIG, NELNS_LOGS);
1207
1208
1209 NLMISC_COMMAND (getViewAS, "send a view and receive an array as result", "<varpath>")
1210 {
1211         string cmd;
1212         for (uint i = 0; i < args.size(); i++)
1213         {
1214                 if (i != 0) cmd += " ";
1215                 cmd += args[i];
1216         }
1217
1218         addRequest (cmd, NULL);
1219
1220         return true;
1221 }
1222
1223 NLMISC_COMMAND (clearRequests, "clear all pending requests", "")
1224 {
1225         if(args.size() != 0) return false;
1226
1227         // for all request, set the NbWaiting to NbReceived, next cleanRequest() will send answer and clear all request
1228         for (uint i = 0 ; i < Requests.size (); i++)
1229         {
1230                 if (Requests[i].NbWaiting <= Requests[i].NbReceived)
1231                 {
1232                         Requests[i].NbWaiting = Requests[i].NbReceived;
1233                 }
1234         }
1235
1236         return true;
1237 }
1238
1239 NLMISC_COMMAND (displayRequests, "display all pending requests", "")
1240 {
1241         if(args.size() != 0) return false;
1242
1243         log.displayNL ("Display %d pending requests", Requests.size ());
1244         for (uint i = 0 ; i < Requests.size (); i++)
1245         {
1246                 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);
1247         }
1248         log.displayNL ("End of display pending requests");
1249
1250         return true;
1251 }
1252
1253 NLMISC_COMMAND (generateAlert, "generate an alert", "<text>")
1254 {
1255         if(args.size() != 1) return false;
1256
1257         sendAdminAlert (args[0].c_str());
1258
1259         return true;
1260 }