aleajactaest/khanat-opennel-code
1
0
Fork 0
mirror of https://port.numenaute.org/aleajactaest/khanat-opennel-code.git synced 2024-12-23 09:28:45 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions
khanat-opennel-code/code/nelns/admin_service/admin_service.cpp

1252 lines
32 KiB
C++
Raw Normal View History

Merge with develop
2014-09-18 19:54:34 +00:00
// NeLNS - MMORPG Framework <http://dev.ryzom.com/projects/nel/>
// Copyright (C) 2010 Winch Gate Property Limited
//
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as
// published by the Free Software Foundation, either version 3 of the
// License, or (at your option) any later version.
//
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program. If not, see <http://www.gnu.org/licenses/>.
#ifdef HAVE_CONFIG_H
#include "config.h"
#endif // HAVE_CONFIG_H
#ifndef NELNS_CONFIG
#define NELNS_CONFIG ""
#endif // NELNS_CONFIG
#ifndef NELNS_LOGS
#define NELNS_LOGS ""
#endif // NELNS_LOGS
#include "nel/misc/types_nl.h"
#include <string>
#include <list>
#ifdef NL_OS_WINDOWS
#include <winsock2.h>
#include <windows.h>
typedef unsigned long ulong;
#endif
#include <mysql.h>
#include <mysql_version.h>
#include "nel/misc/debug.h"
#include "nel/misc/config_file.h"
#include "nel/misc/path.h"
#include "nel/misc/command.h"
#include "nel/net/service.h"
#include "nel/net/varpath.h"
#include "nel/net/email.h"
#include "connection_web.h"
//
// Namespaces
//
using namespace std;
using namespace NLMISC;
using namespace NLNET;
//
// NeL Variables (for config file, etc)
//
// this variable should be used in conjunction with UseExplicitAESRegistration.
// the AS / AES registration process works as follows:
// - aes creates a layer 5 connection to as
// - as gets a serviceUp callback and looks in the database to try to find a match for the AES
// - if the match fails then AS sends a reject message to the AES
// - when the AES receives the reject message they check their UseExplicitAESRegistration flag - if it's set they
// attempt an explicit connection, sending the info required by the AS that would normally come from the database
// - when the AS receives an explicit registration, it verifies the state of the AllowExplicitAESRegistration flag
// and completes the registration work that failed earlier due to the database access failure
CVariable<bool> AllowExplicitAESRegistration("as","AllowExplicitAESRegistration","flag to allow AES services to register explicitly",false,0,true);
// this variable allows one to launch an AS on a machine that doesn't have a database setup
// the functionality of the AS is reduced (particularly in respect to alarms and graphs which are configured via the database)
CVariable<bool> DontUseDataBase("as","DontUseDataBase","if this flag is set calls to the database will be ignored",false,0,true);
//
// Structures
//
struct CRequest
{
CRequest (uint32 id, TSockId from) : Id(id), NbWaiting(0), NbReceived(0), From(from), NbRow(0), NbLines(1)
{
Time = CTime::getSecondsSince1970 ();
}
uint32 Id;
uint NbWaiting;
uint32 NbReceived;
TSockId From;
uint32 Time; // when the request was ask
uint32 NbRow;
uint32 NbLines;
vector<vector<string> > Array; // it's the 2 dimensional array that will be send to the php for variables
vector<string> Log; // this log contains the answer if a command was asked, othewise, Array contains the results
uint32 getVariable(const string &variable)
{
for (uint32 i = 0; i < NbRow; i++)
if (Array[i][0] == variable)
return i;
// need to add the variable
vector<string> NewRow;
NewRow.resize (NbLines);
NewRow[0] = variable;
Array.push_back (NewRow);
return NbRow++;
}
void addLine ()
{
for (uint32 i = 0; i < NbRow; i++)
Array[i].push_back("");
NbLines++;
}
void display ()
{
if (Log.empty())
{
nlinfo ("Display answer array for request %d: %d row %d lines", Id, NbRow, NbLines);
for (uint i = 0; i < NbLines; i++)
{
for (uint j = 0; j < NbRow; j++)
{
nlassert (Array.size () == NbRow);
InfoLog->displayRaw ("%-20s", Array[j][i].c_str());
}
InfoLog->displayRawNL ("");
}
InfoLog->displayRawNL ("End of the array");
}
else
{
nlinfo ("Display the log for request %d: %d lines", Id, Log.size());
for (uint i = 0; i < Log.size(); i++)
{
InfoLog->displayRaw ("%s", Log[i].c_str());
}
InfoLog->displayRawNL ("End of the log");
}
}
};
struct CAdminExecutorService
{
CAdminExecutorService (const string &shard, const string &name, TServiceId sid) : Shard(shard), SId(sid), Name(name) { }
string Shard; /// Name of the shard
TServiceId SId; /// uniq number to identify the AES
string Name; /// name of the admin executor service
vector<uint32> WaitingRequestId; /// contains all request that the server hasn't reply yet
};
typedef list<CAdminExecutorService> TAdminExecutorServices;
typedef list<CAdminExecutorService>::iterator AESIT;
//
// Variables
//
TAdminExecutorServices AdminExecutorServices;
MYSQL *DatabaseConnection = NULL;
vector<CRequest> Requests;
uint32 RequestTimeout = 5; // in second
// cumulate 5 seconds of alert
sint32 AdminAlertAccumlationTime = 5;
//
// Functions
//
AESIT findAES (TServiceId sid, bool asrt = true)
{
AESIT aesit;
for (aesit = AdminExecutorServices.begin(); aesit != AdminExecutorServices.end(); aesit++)
if ((*aesit).SId == sid)
break;
if (asrt)
nlassert (aesit != AdminExecutorServices.end());
return aesit;
}
AESIT findAES (const string &name, bool asrt = true)
{
AESIT aesit;
for (aesit = AdminExecutorServices.begin(); aesit != AdminExecutorServices.end(); aesit++)
if ((*aesit).Name == name)
break;
if (asrt)
nlassert (aesit != AdminExecutorServices.end());
return aesit;
}
//
// SQL helpers
//
MYSQL_RES *sqlCurrentQueryResult = NULL;
MYSQL_ROW sqlQuery (const char *format, ...)
{
if (DontUseDataBase)
return 0;
char *query;
NLMISC_CONVERT_VARGS (query, format, 1024);
if (DatabaseConnection == 0)
{
nlwarning ("MYSQL: mysql_query (%s) failed: DatabaseConnection is 0", query);
return NULL;
}
int ret = mysql_query (DatabaseConnection, query);
if (ret != 0)
{
nlwarning ("MYSQL: mysql_query () failed for query '%s': %s", query, mysql_error(DatabaseConnection));
return 0;
}
sqlCurrentQueryResult = mysql_store_result(DatabaseConnection);
if (sqlCurrentQueryResult == 0)
{
nlwarning ("MYSQL: mysql_store_result () failed for query '%s': %s", query, mysql_error(DatabaseConnection));
return 0;
}
MYSQL_ROW row = mysql_fetch_row(sqlCurrentQueryResult);
if (row == 0)
{
nlwarning ("MYSQL: mysql_fetch_row () failed for query '%s': %s", query, mysql_error(DatabaseConnection));
}
nldebug ("MYSQL: sqlQuery(%s) returns %d rows", query, mysql_num_rows(sqlCurrentQueryResult));
return row;
}
MYSQL_ROW sqlNextRow ()
{
if (DontUseDataBase)
return 0;
if (sqlCurrentQueryResult == 0)
return 0;
return mysql_fetch_row(sqlCurrentQueryResult);
}
void sqlFlushResult()
{
if (DontUseDataBase)
return;
if (sqlCurrentQueryResult == NULL)
return;
mysql_free_result(sqlCurrentQueryResult);
sqlCurrentQueryResult = NULL;
}
//
// Admin functions
//
string Email;
uint32 FirstEmailTime = 0;
void sendAdminAlert (const char *format, ...)
{
char *text;
NLMISC_CONVERT_VARGS (text, format, 4096);
if (AdminAlertAccumlationTime == -1)
{
// we don't send email so just display a warning
nlwarning ("ALERT: %s", text);
}
else
{
if(Email.empty() && FirstEmailTime == 0)
{
Email += text;
FirstEmailTime = CTime::getSecondsSince1970();
}
else
{
Email += "\n";
Email += text;
}
nldebug ("ALERT: pushing email into queue: %s", text);
}
}
void updateSendAdminAlert ()
{
if(!Email.empty() && FirstEmailTime != 0 && AdminAlertAccumlationTime >=0 && CTime::getSecondsSince1970() > FirstEmailTime + AdminAlertAccumlationTime)
{
vector<string> lines;
explode (Email, string("\n"), lines, true);
if (!lines.empty())
{
if (IService::getInstance()->ConfigFile.exists("SysLogPath") && IService::getInstance()->ConfigFile.exists("SysLogParams"))
{
// syslog
string param;
if (lines.size() > 1)
{
param = "Multiple problems, first is: ";
}
param += lines[0];
string res = toString(IService::getInstance()->ConfigFile.getVar("SysLogParams").asString().c_str(), param.c_str());
launchProgram(IService::getInstance()->ConfigFile.getVar("SysLogPath").asString(), res);
}
if (IService::getInstance()->ConfigFile.exists("AdminEmail"))
{
// email
string subject;
if (lines.size() == 1)
{
subject = lines[0];
}
else
{
subject = "Multiple problems";
}
std::string from;
if(IService::getInstance()->ConfigFile.exists("AdminEmailFrom"))
from = IService::getInstance()->ConfigFile.getVar("AdminEmailFrom").asString();
CConfigFile::CVar &var = IService::getInstance()->ConfigFile.getVar("AdminEmail");
for (uint i = 0; i < var.size(); i++)
{
if (!sendEmail ("", from, var.asString(i), subject, Email))
{
nlwarning ("Can't send email to '%s'", var.asString(i).c_str());
}
else
{
nlinfo ("ALERT: Sent email to admin %s the subject: %s", var.asString(i).c_str(), subject.c_str());
}
}
}
}
Email = "";
FirstEmailTime = 0;
}
}
static void cbAdminEmail (CMessage &msgin, const std::string &serviceName, TServiceId sid)
{
string str;
msgin.serial(str);
sendAdminAlert (str.c_str());
}
static void cbGraphUpdate (CMessage &msgin, const std::string &serviceName, TServiceId sid)
{
uint32 CurrentTime;
msgin.serial (CurrentTime);
while (msgin.getPos() < (sint32)msgin.length())
{
string var, service;
sint32 val;
msgin.serial (service, var, val);
AESIT aesit = findAES (sid);
string shard, server;
shard = (*aesit).Shard;
server = (*aesit).Name;
if (!shard.empty() && !server.empty() && !service.empty() && !var.empty())
{
string path = CPath::standardizePath (IService::getInstance()->ConfigFile.getVar("RRDVarPath").asString());
string rrdfilename = path + shard+"."+server+"."+service+"."+var+".rrd";
string arg;
if (!NLMISC::CFile::fileExists(rrdfilename))
{
MYSQL_ROW row = sqlQuery ("select graph_update from variable where path like '%%%s' and graph_update!=0", var.c_str());
if (row != NULL)
{
uint32 freq = atoi(row[0]);
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";
launchProgram(IService::getInstance()->ConfigFile.getVar("RRDToolPath").asString(), arg);
}
else
{
nlwarning ("Can't create the rrd because no graph_update in database");
}
sqlFlushResult();
}
arg = "update " + rrdfilename + " " + toString (CurrentTime) + ":" + toString(val);
launchProgram(IService::getInstance()->ConfigFile.getVar("RRDToolPath").asString(), arg);
}
else
{
nlwarning ("Shard server service var val is empty");
}
}
}
//
// Request functions
//
uint32 newRequest (TSockId from)
{
static uint32 NextId = 5461231;
Requests.push_back (CRequest(NextId, from));
return NextId++;
}
void addRequestWaitingNb (uint32 rid)
{
for (uint i = 0 ; i < Requests.size (); i++)
{
if (Requests[i].Id == rid)
{
Requests[i].NbWaiting++;
Requests[i].Time = CTime::getSecondsSince1970 ();
return;
}
}
nlwarning ("REQUEST: Received an answer from an unknown resquest %d (perhaps due to a AS timeout)", rid);
}
void subRequestWaitingNb (uint32 rid)
{
for (uint i = 0 ; i < Requests.size (); i++)
{
if (Requests[i].Id == rid)
{
Requests[i].NbWaiting--;
return;
}
}
nlwarning ("REQUEST: Received an answer from an unknown resquest %d (perhaps due to a AS timeout)", rid);
}
void addRequestReceived (uint32 rid)
{
for (uint i = 0 ; i < Requests.size (); i++)
{
if (Requests[i].Id == rid)
{
Requests[i].NbReceived++;
return;
}
}
nlwarning ("REQUEST: Received an answer from an unknown resquest %d (perhaps due to a AS timeout)", rid);
}
void addRequestAnswer (uint32 rid, const vector<string> &variables, const vector<string> &values)
{
for (uint i = 0 ; i < Requests.size (); i++)
{
Requests[i].addLine ();
if (Requests[i].Id == rid)
{
if (!variables.empty() && variables[0]=="__log")
{
nlassert (variables.size() == 1);
for (uint j = 0; j < values.size(); j++)
{
Requests[i].Log.push_back (values[j]);
}
}
else
{
nlassert (variables.size() == values.size ());
for (uint j = 0; j < variables.size(); j++)
{
uint32 pos = Requests[i].getVariable (variables[j]);
Requests[i].Array[pos][Requests[i].NbLines-1] = values[j];
}
}
return;
}
}
nlwarning ("REQUEST: Received an answer from an unknown resquest %d (perhaps due to a AS timeout)", rid);
}
bool emptyRequest (uint32 rid)
{
for (uint i = 0 ; i < Requests.size (); i++)
{
if (Requests[i].Id == rid && Requests[i].NbWaiting != 0)
{
return false;
}
}
return true;
}
void cleanRequest ()
{
uint32 currentTime = CTime::getSecondsSince1970 ();
bool timeout;
for (uint i = 0 ; i < Requests.size ();)
{
// the AES doesn't answer quickly
timeout = (currentTime >= Requests[i].Time+RequestTimeout);
if (Requests[i].NbWaiting <= Requests[i].NbReceived || timeout)
{
// the request is over, send to the php
string str;
if (timeout)
{
nlwarning ("REQUEST: Request %d timeouted, only %d on %d services have replied", Requests[i].Id, Requests[i].NbReceived, Requests[i].NbWaiting);
}
if (Requests[i].Log.empty())
{
if (Requests[i].NbRow == 0 && timeout)
{
str = "1 ((TIMEOUT))";
}
else
{
str = toString(Requests[i].NbRow) + " ";
for (uint k = 0; k < Requests[i].NbLines; k++)
{
for (uint j = 0; j < Requests[i].NbRow; j++)
{
nlassert (Requests[i].Array.size () == Requests[i].NbRow);
if (Requests[i].Array[j][k].empty ())
str += "??? ";
else
{
str += Requests[i].Array[j][k];
if (timeout)
str += "((TIMEOUT))";
str += " ";
}
}
}
}
}
else
{
for (uint k = 0; k < Requests[i].Log.size(); k++)
{
str += Requests[i].Log[k];
if (timeout)
str += "((TIMEOUT))";
}
}
sendString (Requests[i].From, str);
// set to 0 to erase it
Requests[i].NbWaiting = 0;
}
if (Requests[i].NbWaiting == 0)
{
Requests.erase (Requests.begin ()+i);
}
else
{
i++;
}
}
}
//
// SQL functions
//
void sqlInit ()
{
if (DontUseDataBase)
return;
MYSQL *db = mysql_init(NULL);
if(db == NULL)
{
nlerror ("mysql_init() failed");
}
my_bool opt = true;
if (mysql_options (db, MYSQL_OPT_RECONNECT, &opt))
{
mysql_close(db);
DatabaseConnection = 0;
nlerror("mysql_options() failed for database connection to '%s'", IService::getInstance()->ConfigFile.getVar("DatabaseHost").asString().c_str());
return;
}
DatabaseConnection = mysql_real_connect(db,
IService::getInstance()->ConfigFile.getVar("DatabaseHost").asString().c_str(),
IService::getInstance()->ConfigFile.getVar("DatabaseLogin").asString().c_str(),
IService::getInstance()->ConfigFile.getVar("DatabasePassword").asString().c_str(),
IService::getInstance()->ConfigFile.getVar("DatabaseName").asString().c_str(),
0,NULL,0);
if (DatabaseConnection == NULL || DatabaseConnection != db)
{
nlerror ("mysql_real_connect() failed to '%s' with login '%s' and database name '%s' with %s",
IService::getInstance()->ConfigFile.getVar("DatabaseHost").asString().c_str(),
IService::getInstance()->ConfigFile.getVar("DatabaseLogin").asString().c_str(),
IService::getInstance()->ConfigFile.getVar("DatabaseName").asString().c_str(),
(IService::getInstance()->ConfigFile.getVar("DatabasePassword").asString().empty()?"empty password":"password")
);
}
#if MYSQL_VERSION_ID < 50019
opt = true;
if (mysql_options (DatabaseConnection, MYSQL_OPT_RECONNECT, &opt))
{
mysql_close(db);
DatabaseConnection = 0;
nlerror("mysql_options() failed for database connection to '%s'", IService::getInstance()->ConfigFile.getVar("DatabaseHost").asString().c_str());
return;
}
#endif
}
////////////////////////////////////////////////////////////////////////////////////////////////////////
////////////////////////////////////////////////////////////////////////////////////////////////////////
////////////////// CONNECTION TO THE AES ///////////////////////////////////////////////////////////////
////////////////////////////////////////////////////////////////////////////////////////////////////////
////////////////////////////////////////////////////////////////////////////////////////////////////////
void sendAESInformation (TServiceId sid)
{
AESIT aesit = findAES (sid);
vector<string> information;
CMessage msgout("AES_INFO");
//
// send services that should be running on this AES
//
information.clear ();
MYSQL_ROW row = sqlQuery ("select name from service where server='%s'", (*aesit).Name.c_str());
while (row != NULL)
{
string service = row[0];
nlinfo ("Adding '%s' in registered services to AES-%hu", row[0], sid.get());
information.push_back (service);
row = sqlNextRow ();
}
sqlFlushResult();
msgout.serialCont (information);
//
// send variable alarms for services that should running on this AES
//
information.clear ();
row = sqlQuery ("select path, error_bound, alarm_order from variable where error_bound!=-1");
while (row != NULL)
{
nlinfo ("Adding '%s' '%s' '%s' in alarm to AES-%hu", row[0], row[1], row[2], sid.get());
information.push_back (row[0]);
information.push_back (row[1]);
information.push_back (row[2]);
row = sqlNextRow ();
}
sqlFlushResult();
msgout.serialCont (information);
//
// send graph update for services that should running on this AES
//
information.clear ();
row = sqlQuery ("select path, graph_update from variable where graph_update!=0");
while (row != NULL)
{
CVarPath varpath (row[0]);
for(uint i = 0; i < varpath.Destination.size(); i++)
{
string a = varpath.Destination[i].first, b = (*aesit).Shard;
if(varpath.Destination[i].first == "*" || varpath.Destination[i].first == (*aesit).Shard)
{
CVarPath varpath2 (varpath.Destination[i].second);
for(uint j = 0; j < varpath2.Destination.size(); j++)
{
string c = varpath2.Destination[j].first, d = (*aesit).Name;
if(varpath2.Destination[j].first == "*" || varpath2.Destination[j].first == (*aesit).Name)
{
nlinfo ("Adding '%s' '%s' in graph to AES-%hu", row[0], row[1], sid.get());
information.push_back (row[0]);
information.push_back (row[1]);
}
}
}
}
row = sqlNextRow ();
}
sqlFlushResult();
msgout.serialCont (information);
nlinfo ("Sending all information about %s AES-%hu (hostedservices, alarms,grapupdate)", (*aesit).Name.c_str(), (*aesit).SId.get());
CUnifiedNetwork::getInstance ()->send (sid, msgout);
}
void rejectAES(TServiceId sid, const string &res)
{
CMessage msgout("REJECTED");
msgout.serial ((string &)res);
CUnifiedNetwork::getInstance ()->send (sid, msgout);
}
// i'm connected to a new admin executor service
static void cbNewAESConnection (const std::string &serviceName, TServiceId sid, void *arg)
{
TSockId from;
CCallbackNetBase *cnb = CUnifiedNetwork::getInstance ()->getNetBase (sid, from);
const CInetAddress &ia = cnb->hostAddress (from);
AESIT aesit = findAES (sid, false);
if (aesit != AdminExecutorServices.end ())
{
nlwarning ("Connection of an AES that is already in the list (%s)", ia.asString ().c_str ());
rejectAES (sid, "This AES is already in the AS list");
return;
}
MYSQL_ROW row = sqlQuery ("select name from server where address='%s'", ia.ipAddress().c_str());
if (row == NULL)
{
if (!AllowExplicitAESRegistration)
{
nlwarning ("Connection of an AES that is not in database server list (%s)", ia.asString ().c_str ());
}
else
{
nlinfo ("Rejecting auto-connection of an AES (%s) - this should provke explicitly reconnect", ia.asString ().c_str ());
}
rejectAES (sid, "This AES is not registered in the database");
sqlFlushResult();
return;
}
string server = row[0];
sqlFlushResult();
row = sqlQuery ("select shard from service where server='%s'", server.c_str());
if (row == NULL)
{
nlwarning ("Connection of an AES that is not in database server list (%s)", ia.asString ().c_str ());
rejectAES (sid, "This AES is not registered in the database");
sqlFlushResult();
return;
}
string shard = row[0];
sqlFlushResult();
AdminExecutorServices.push_back (CAdminExecutorService(shard, server, sid));
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());
// send him services that should run on this server
sendAESInformation (sid);
}
// i'm disconnected from an admin executor service
static void cbNewAESDisconnection (const std::string &serviceName, TServiceId sid, void *arg)
{
TSockId from;
CCallbackNetBase *cnb = CUnifiedNetwork::getInstance ()->getNetBase (sid, from);
const CInetAddress &ia = cnb->hostAddress (from);
AESIT aesit = findAES (sid, false);
if (aesit == AdminExecutorServices.end ())
{
nlwarning ("Disconnection of %s-%hu that is not in my list (%s)", serviceName.c_str (), sid.get(), ia.asString ().c_str ());
return;
}
nlinfo ("%s-%hu, shard name %s, disconnected and removed from the list", serviceName.c_str(), sid.get(), (*aesit).Name.c_str ());
// we need to remove pending request
for(uint i = 0; i < (*aesit).WaitingRequestId.size (); i++)
{
subRequestWaitingNb ((*aesit).WaitingRequestId[i]);
}
AdminExecutorServices.erase (aesit);
}
// we receive an explicit registration message from an AES
void cbRegisterAES(CMessage &msgin, const std::string &serviceName, TServiceId sid)
{
if (!AllowExplicitAESRegistration)
{
nlwarning("Ignoring attempted AES registration because AllowExplicitAESRegistration==false");
return;
}
string server;
string shard;
try
{
msgin.serial(server);
msgin.serial(shard);
}
catch(...)
{
nlwarning("Ignoring attempted AES registration due to execption during message decoding");
return;
}
AdminExecutorServices.push_back (CAdminExecutorService(shard, server, sid));
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());
// send him services that should run on this server
sendAESInformation (sid);
}
static void cbView (CMessage &msgin, const std::string &serviceName, TServiceId sid)
{
uint32 rid;
msgin.serial (rid);
AESIT aesit = findAES (sid);
for (uint i = 0; i < (*aesit).WaitingRequestId.size();)
{
if ((*aesit).WaitingRequestId[i] == rid)
{
(*aesit).WaitingRequestId.erase ((*aesit).WaitingRequestId.begin ()+i);
}
else
{
i++;
}
}
MYSQL_ROW row = sqlQuery ("select distinct shard from service where server='%s'", (*aesit).Name.c_str ());
// shard name is find using the "service" table, so, if there s no shard name in it, it returns ???
string shardName;
if (row != NULL) shardName = row[0];
else shardName = DontUseDataBase? aesit->Shard: "???";
vector<string> vara, vala;
while ((uint32)msgin.getPos() < msgin.length())
{
vara.clear ();
vala.clear ();
// adding default row
vara.push_back ("shard");
vara.push_back ("server");
vala.push_back (shardName);
vala.push_back ((*aesit).Name);
uint32 i, nb;
string var, val;
msgin.serial (nb);
for (i = 0; i < nb; i++)
{
msgin.serial (var);
if (var == "__log")
{
vara.clear ();
vala.clear ();
}
vara.push_back (var);
}
if (vara.size() > 0 && vara[0] == "__log")
vala.push_back ("----- Result from Shard "+shardName+" Server "+(*aesit).Name+"\n");
msgin.serial (nb);
for (i = 0; i < nb; i++)
{
msgin.serial (val);
vala.push_back (val);
}
addRequestAnswer (rid, vara, vala);
}
sqlFlushResult();
// inc the NbReceived counter
addRequestReceived (rid);
}
TUnifiedCallbackItem CallbackArray[] =
{
{ "REGISTER_AES", cbRegisterAES },
{ "VIEW", cbView },
{ "ADMIN_EMAIL", cbAdminEmail },
{ "GRAPH_UPDATE", cbGraphUpdate },
};
////////////////////////////////////////////////////////////////////////////////////////////////////////
////////////////////////////////////////////////////////////////////////////////////////////////////////
////////////////// CONNECTION TO THE CLIENT ////////////////////////////////////////////////////////////
////////////////////////////////////////////////////////////////////////////////////////////////////////
////////////////////////////////////////////////////////////////////////////////////////////////////////
void addRequest (const string &rawvarpath, TSockId from)
{
nlinfo ("addRequest from %s: '%s'", from->asString ().c_str (), rawvarpath.c_str ());
if(rawvarpath.empty ())
{
// send an empty string to say to php that there's nothing
string str;
sendString (from, str);
}
//
// special cases
//
if(rawvarpath == "reload")
{
// it means the we have to resend the list of services managed by AES from the mysql tables
for (AESIT aesit = AdminExecutorServices.begin(); aesit != AdminExecutorServices.end(); aesit++)
{
sendAESInformation ((*aesit).SId);
}
// send an empty string to say to php that there's nothing
string str;
sendString (from, str);
return;
}
//
// normal cases
//
CVarPath varpath (rawvarpath);
uint32 rid = newRequest (from);
for (uint i = 0; i < varpath.Destination.size (); i++)
{
string shard = varpath.Destination[i].first;
CVarPath subvarpath (varpath.Destination[i].second);
for (uint j = 0; j < subvarpath.Destination.size (); j++)
{
string server = subvarpath.Destination[j].first;
if (shard == "*" && server == "*")
{
// Send the request to all online servers of all online shards
AESIT aesit;
for (aesit = AdminExecutorServices.begin(); aesit != AdminExecutorServices.end(); aesit++)
{
addRequestWaitingNb (rid);
(*aesit).WaitingRequestId.push_back (rid);
CMessage msgout("AES_GET_VIEW");
msgout.serial (rid);
msgout.serial (subvarpath.Destination[j].second);
CUnifiedNetwork::getInstance ()->send ((*aesit).SId, msgout);
nlinfo ("REQUEST: Sent view '%s' to shard name %s 'AES-%hu'", subvarpath.Destination[j].second.c_str(), (*aesit).Name.c_str(), (*aesit).SId.get());
}
}
else if (shard == "*" && server == "#")
{
// Select all shard all server including offline one
MYSQL_ROW row = sqlQuery ("select distinct server, shard from service");
while (row != NULL)
{
AESIT aesit = findAES (row[0], false);
if (aesit != AdminExecutorServices.end())
{
addRequestWaitingNb (rid);
(*aesit).WaitingRequestId.push_back (rid);
CMessage msgout("AES_GET_VIEW");
msgout.serial (rid);
msgout.serial (subvarpath.Destination[j].second);
CUnifiedNetwork::getInstance ()->send ((*aesit).SId, msgout);
nlinfo ("REQUEST: Sent view '%s' to shard name %s 'AES-%hu'", subvarpath.Destination[j].second.c_str(), (*aesit).Name.c_str(), (*aesit).SId.get());
}
else if (server == "#")
{
vector<string> vara, vala;
// adding default row
vara.push_back ("shard");
vala.push_back (row[1]);
vara.push_back ("server");
vala.push_back (row[0]);
vara.push_back ("service");
vala.push_back ("AES");
vara.push_back ("State");
vala.push_back ("Offline");
addRequestAnswer (rid, vara, vala);
}
row = sqlNextRow ();
}
sqlFlushResult();
}
else if (server == "*" || server == "#")
{
// Send the request to all online server of a specific shard
MYSQL_ROW row = sqlQuery ("select distinct server from service where shard='%s'", shard.c_str ());
while (row != NULL)
{
AESIT aesit = findAES (row[0], false);
if (aesit != AdminExecutorServices.end())
{
addRequestWaitingNb (rid);
(*aesit).WaitingRequestId.push_back (rid);
CMessage msgout("AES_GET_VIEW");
msgout.serial (rid);
msgout.serial (subvarpath.Destination[j].second);
CUnifiedNetwork::getInstance ()->send ((*aesit).SId, msgout);
nlinfo ("REQUEST: Sent view '%s' to shard name %s 'AES-%hu'", subvarpath.Destination[j].second.c_str(), (*aesit).Name.c_str(), (*aesit).SId.get());
}
else if (server == "#")
{
vector<string> vara, vala;
// adding default row
vara.push_back ("shard");
vala.push_back (shard);
vara.push_back ("server");
vala.push_back (row[0]);
vara.push_back ("service");
vala.push_back ("AES");
vara.push_back ("State");
vala.push_back ("Offline");
addRequestAnswer (rid, vara, vala);
}
row = sqlNextRow ();
}
sqlFlushResult();
}
else
{
AESIT aesit = findAES (server, false);
if (aesit != AdminExecutorServices.end())
{
addRequestWaitingNb (rid);
(*aesit).WaitingRequestId.push_back (rid);
CMessage msgout("AES_GET_VIEW");
msgout.serial (rid);
msgout.serial (subvarpath.Destination[j].second);
CUnifiedNetwork::getInstance ()->send ((*aesit).SId, msgout);
nlinfo ("REQUEST: Sent view '%s' to shard name %s 'AES-%hu'", subvarpath.Destination[j].second.c_str(), (*aesit).Name.c_str(), (*aesit).SId.get());
}
else
{
nlwarning ("Server %s is not found in the list", server.c_str ());
}
}
}
}
}
static void varRequestTimeout(CConfigFile::CVar &var)
{
RequestTimeout = var.asInt();
nlinfo ("Request timeout is now after %d seconds", RequestTimeout);
}
static void varAdminAlertAccumlationTime (CConfigFile::CVar &var)
{
AdminAlertAccumlationTime = var.asInt();
}
////////////////////////////////////////////////////////////////////////////////////////////////////////
////////////////////////////////////////////////////////////////////////////////////////////////////////
////////////////// SERVICE IMPLEMENTATION //////////////////////////////////////////////////////////////
////////////////////////////////////////////////////////////////////////////////////////////////////////
////////////////////////////////////////////////////////////////////////////////////////////////////////
class CAdminService : public IService
{
public:
/// Init the service, load the universal time.
void init ()
{
setDefaultEmailParams (ConfigFile.getVar ("SMTPServer").asString (), ConfigFile.getVar("DefaultEmailFrom").asString(), "");
sqlInit ();
connectionWebInit ();
//CVarPath toto ("[toto");
//CVarPath toto ("*.*.*.*");
//CVarPath toto ("[srv1,srv2].*.*.*");
//CVarPath toto ("[svr1.svc1,srv2.svc2].*.*");
//CVarPath toto ("[svr1.[svc1,svc2].*.var1,srv2.svc2.fe*.var2].toto");
//CVarPath toto ("[svr1.svc1.*.toto,srv2.svc2.*.tata]");
CUnifiedNetwork::getInstance ()->setServiceUpCallback ("AES", cbNewAESConnection);
CUnifiedNetwork::getInstance ()->setServiceDownCallback ("AES", cbNewAESDisconnection);
varRequestTimeout (ConfigFile.getVar ("RequestTimeout"));
ConfigFile.setCallback("RequestTimeout", &varRequestTimeout);
varAdminAlertAccumlationTime (ConfigFile.getVar ("AdminAlertAccumlationTime"));
ConfigFile.setCallback("AdmimAlertAccumlationTime", &varAdminAlertAccumlationTime);
}
bool update ()
{
cleanRequest ();
connectionWebUpdate ();
updateSendAdminAlert ();
return true;
}
void release ()
{
connectionWebRelease ();
}
};
/// Admin Service
NLNET_SERVICE_MAIN (CAdminService, "AS", "admin_service", 49996, CallbackArray, NELNS_CONFIG, NELNS_LOGS);
NLMISC_COMMAND (getViewAS, "send a view and receive an array as result", "<varpath>")
{
string cmd;
for (uint i = 0; i < args.size(); i++)
{
if (i != 0) cmd += " ";
cmd += args[i];
}
addRequest (cmd, NULL);
return true;
}
NLMISC_COMMAND (clearRequests, "clear all pending requests", "")
{
if(args.size() != 0) return false;
// for all request, set the NbWaiting to NbReceived, next cleanRequest() will send answer and clear all request
for (uint i = 0 ; i < Requests.size (); i++)
{
if (Requests[i].NbWaiting <= Requests[i].NbReceived)
{
Requests[i].NbWaiting = Requests[i].NbReceived;
}
}
return true;
}
NLMISC_COMMAND (displayRequests, "display all pending requests", "")
{
if(args.size() != 0) return false;
log.displayNL ("Display %d pending requests", Requests.size ());
for (uint i = 0 ; i < Requests.size (); i++)
{
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);
}
log.displayNL ("End of display pending requests");
return true;
}
NLMISC_COMMAND (generateAlert, "generate an alert", "<text>")
{
if(args.size() != 1) return false;
sendAdminAlert (args[0].c_str());
return true;
}
Reference in a new issue Copy permalink