granicus.if.org Git - icinga2/blob - lib/db_ido_mysql/idomysqlconnection.cpp

   1 /******************************************************************************
   2  * Icinga 2                                                                   *
   3  * Copyright (C) 2012-2017 Icinga Development Team (https://www.icinga.com/)  *
   4  *                                                                            *
   5  * This program is free software; you can redistribute it and/or              *
   6  * modify it under the terms of the GNU General Public License                *
   7  * as published by the Free Software Foundation; either version 2             *
   8  * of the License, or (at your option) any later version.                     *
   9  *                                                                            *
  10  * This program is distributed in the hope that it will be useful,            *
  11  * but WITHOUT ANY WARRANTY; without even the implied warranty of             *
  12  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the              *
  13  * GNU General Public License for more details.                               *
  14  *                                                                            *
  15  * You should have received a copy of the GNU General Public License          *
  16  * along with this program; if not, write to the Free Software Foundation     *
  17  * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301, USA.             *
  18  ******************************************************************************/
  19
  20 #include "db_ido_mysql/idomysqlconnection.hpp"
  21 #include "db_ido_mysql/idomysqlconnection.tcpp"
  22 #include "icinga/perfdatavalue.hpp"
  23 #include "db_ido/dbtype.hpp"
  24 #include "db_ido/dbvalue.hpp"
  25 #include "base/logger.hpp"
  26 #include "base/objectlock.hpp"
  27 #include "base/convert.hpp"
  28 #include "base/utility.hpp"
  29 #include "base/application.hpp"
  30 #include "base/configtype.hpp"
  31 #include "base/exception.hpp"
  32 #include "base/statsfunction.hpp"
  33 #include <boost/tuple/tuple.hpp>
  34
  35 using namespace icinga;
  36
  37 REGISTER_TYPE(IdoMysqlConnection);
  38 REGISTER_STATSFUNCTION(IdoMysqlConnection, &IdoMysqlConnection::StatsFunc);
  39
  40 IdoMysqlConnection::IdoMysqlConnection(void)
  41         : m_QueryQueue(10000000)
  42 { }
  43
  44 void IdoMysqlConnection::OnConfigLoaded(void)
  45 {
  46         ObjectImpl<IdoMysqlConnection>::OnConfigLoaded();
  47
  48         m_QueryQueue.SetName("IdoMysqlConnection, " + GetName());
  49 }
  50
  51 void IdoMysqlConnection::StatsFunc(const Dictionary::Ptr& status, const Array::Ptr& perfdata)
  52 {
  53         Dictionary::Ptr nodes = new Dictionary();
  54
  55         for (const IdoMysqlConnection::Ptr& idomysqlconnection : ConfigType::GetObjectsByType<IdoMysqlConnection>()) {
  56                 size_t items = idomysqlconnection->m_QueryQueue.GetLength();
  57
  58                 Dictionary::Ptr stats = new Dictionary();
  59                 stats->Set("version", idomysqlconnection->GetSchemaVersion());
  60                 stats->Set("instance_name", idomysqlconnection->GetInstanceName());
  61                 stats->Set("connected", idomysqlconnection->GetConnected());
  62                 stats->Set("query_queue_items", items);
  63
  64                 nodes->Set(idomysqlconnection->GetName(), stats);
  65
  66                 perfdata->Add(new PerfdataValue("idomysqlconnection_" + idomysqlconnection->GetName() + "_queries_rate", idomysqlconnection->GetQueryCount(60) / 60.0));
  67                 perfdata->Add(new PerfdataValue("idomysqlconnection_" + idomysqlconnection->GetName() + "_queries_1min", idomysqlconnection->GetQueryCount(60)));
  68                 perfdata->Add(new PerfdataValue("idomysqlconnection_" + idomysqlconnection->GetName() + "_queries_5mins", idomysqlconnection->GetQueryCount(5 * 60)));
  69                 perfdata->Add(new PerfdataValue("idomysqlconnection_" + idomysqlconnection->GetName() + "_queries_15mins", idomysqlconnection->GetQueryCount(15 * 60)));
  70                 perfdata->Add(new PerfdataValue("idomysqlconnection_" + idomysqlconnection->GetName() + "_query_queue_items", items));
  71         }
  72
  73         status->Set("idomysqlconnection", nodes);
  74 }
  75
  76 void IdoMysqlConnection::Resume(void)
  77 {
  78         DbConnection::Resume();
  79
  80         Log(LogInformation, "IdoMysqlConnection")
  81             << "'" << GetName() << "' resumed.";
  82
  83         SetConnected(false);
  84
  85         m_QueryQueue.SetExceptionCallback(boost::bind(&IdoMysqlConnection::ExceptionHandler, this, _1));
  86
  87         m_TxTimer = new Timer();
  88         m_TxTimer->SetInterval(1);
  89         m_TxTimer->OnTimerExpired.connect(boost::bind(&IdoMysqlConnection::TxTimerHandler, this));
  90         m_TxTimer->Start();
  91
  92         m_ReconnectTimer = new Timer();
  93         m_ReconnectTimer->SetInterval(10);
  94         m_ReconnectTimer->OnTimerExpired.connect(boost::bind(&IdoMysqlConnection::ReconnectTimerHandler, this));
  95         m_ReconnectTimer->Start();
  96         m_ReconnectTimer->Reschedule(0);
  97
  98         ASSERT(mysql_thread_safe());
  99 }
 100
 101 void IdoMysqlConnection::Pause(void)
 102 {
 103         Log(LogInformation, "IdoMysqlConnection")
 104             << "'" << GetName() << "' paused.";
 105
 106         m_ReconnectTimer.reset();
 107
 108         DbConnection::Pause();
 109
 110 #ifdef I2_DEBUG /* I2_DEBUG */
 111         Log(LogDebug, "IdoMysqlConnection")
 112             << "Rescheduling disconnect task.";
 113 #endif /* I2_DEBUG */
 114
 115         m_QueryQueue.Enqueue(boost::bind(&IdoMysqlConnection::Disconnect, this), PriorityHigh);
 116         m_QueryQueue.Join();
 117 }
 118
 119 void IdoMysqlConnection::ExceptionHandler(boost::exception_ptr exp)
 120 {
 121         Log(LogCritical, "IdoMysqlConnection", "Exception during database operation: Verify that your database is operational!");
 122
 123         Log(LogDebug, "IdoMysqlConnection")
 124             << "Exception during database operation: " << DiagnosticInformation(exp);
 125
 126         if (GetConnected()) {
 127                 mysql_close(&m_Connection);
 128
 129                 SetConnected(false);
 130         }
 131 }
 132
 133 void IdoMysqlConnection::AssertOnWorkQueue(void)
 134 {
 135         ASSERT(m_QueryQueue.IsWorkerThread());
 136 }
 137
 138 void IdoMysqlConnection::Disconnect(void)
 139 {
 140         AssertOnWorkQueue();
 141
 142         if (!GetConnected())
 143                 return;
 144
 145         Query("COMMIT");
 146         mysql_close(&m_Connection);
 147
 148         SetConnected(false);
 149 }
 150
 151 void IdoMysqlConnection::TxTimerHandler(void)
 152 {
 153         NewTransaction();
 154 }
 155
 156 void IdoMysqlConnection::NewTransaction(void)
 157 {
 158 #ifdef I2_DEBUG /* I2_DEBUG */
 159         Log(LogDebug, "IdoMysqlConnection")
 160             << "Scheduling new transaction and finishing async queries.";
 161 #endif /* I2_DEBUG */
 162
 163         m_QueryQueue.Enqueue(boost::bind(&IdoMysqlConnection::InternalNewTransaction, this), PriorityHigh);
 164         m_QueryQueue.Enqueue(boost::bind(&IdoMysqlConnection::FinishAsyncQueries, this), PriorityHigh);
 165 }
 166
 167 void IdoMysqlConnection::InternalNewTransaction(void)
 168 {
 169         AssertOnWorkQueue();
 170
 171         if (!GetConnected())
 172                 return;
 173
 174         AsyncQuery("COMMIT");
 175         AsyncQuery("BEGIN");
 176 }
 177
 178 void IdoMysqlConnection::ReconnectTimerHandler(void)
 179 {
 180 #ifdef I2_DEBUG /* I2_DEBUG */
 181         Log(LogDebug, "IdoMysqlConnection")
 182             << "Scheduling reconnect task.";
 183 #endif /* I2_DEBUG */
 184
 185         m_QueryQueue.Enqueue(boost::bind(&IdoMysqlConnection::Reconnect, this), PriorityLow);
 186 }
 187
 188 void IdoMysqlConnection::Reconnect(void)
 189 {
 190         AssertOnWorkQueue();
 191
 192         if (!IsActive())
 193                 return;
 194
 195         CONTEXT("Reconnecting to MySQL IDO database '" + GetName() + "'");
 196
 197         double startTime = Utility::GetTime();
 198
 199         SetShouldConnect(true);
 200
 201         bool reconnect = false;
 202
 203         if (GetConnected()) {
 204                 /* Check if we're really still connected */
 205                 if (mysql_ping(&m_Connection) == 0)
 206                         return;
 207
 208                 mysql_close(&m_Connection);
 209                 SetConnected(false);
 210                 reconnect = true;
 211         }
 212
 213         ClearIDCache();
 214
 215         String ihost, isocket_path, iuser, ipasswd, idb;
 216         String isslKey, isslCert, isslCa, isslCaPath, isslCipher;
 217         const char *host, *socket_path, *user , *passwd, *db;
 218         const char *sslKey, *sslCert, *sslCa, *sslCaPath, *sslCipher;
 219         bool enableSsl;
 220         long port;
 221
 222         ihost = GetHost();
 223         isocket_path = GetSocketPath();
 224         iuser = GetUser();
 225         ipasswd = GetPassword();
 226         idb = GetDatabase();
 227
 228         enableSsl = GetEnableSsl();
 229         isslKey = GetSslKey();
 230         isslCert = GetSslCert();
 231         isslCa = GetSslCa();
 232         isslCaPath = GetSslCapath();
 233         isslCipher = GetSslCipher();
 234
 235         host = (!ihost.IsEmpty()) ? ihost.CStr() : NULL;
 236         port = GetPort();
 237         socket_path = (!isocket_path.IsEmpty()) ? isocket_path.CStr() : NULL;
 238         user = (!iuser.IsEmpty()) ? iuser.CStr() : NULL;
 239         passwd = (!ipasswd.IsEmpty()) ? ipasswd.CStr() : NULL;
 240         db = (!idb.IsEmpty()) ? idb.CStr() : NULL;
 241
 242         sslKey = (!isslKey.IsEmpty()) ? isslKey.CStr() : NULL;
 243         sslCert = (!isslCert.IsEmpty()) ? isslCert.CStr() : NULL;
 244         sslCa = (!isslCa.IsEmpty()) ? isslCa.CStr() : NULL;
 245         sslCaPath = (!isslCaPath.IsEmpty()) ? isslCaPath.CStr() : NULL;
 246         sslCipher = (!isslCipher.IsEmpty()) ? isslCipher.CStr() : NULL;
 247
 248         /* connection */
 249         if (!mysql_init(&m_Connection)) {
 250                 Log(LogCritical, "IdoMysqlConnection")
 251                     << "mysql_init() failed: \"" << mysql_error(&m_Connection) << "\"";
 252
 253                 BOOST_THROW_EXCEPTION(std::bad_alloc());
 254         }
 255
 256         if (enableSsl)
 257                 mysql_ssl_set(&m_Connection, sslKey, sslCert, sslCa, sslCaPath, sslCipher);
 258
 259         if (!mysql_real_connect(&m_Connection, host, user, passwd, db, port, socket_path, CLIENT_FOUND_ROWS | CLIENT_MULTI_STATEMENTS)) {
 260                 Log(LogCritical, "IdoMysqlConnection")
 261                     << "Connection to database '" << db << "' with user '" << user << "' on '" << host << ":" << port
 262                     << "' " << (enableSsl ? "(SSL enabled) " : "") << "failed: \"" << mysql_error(&m_Connection) << "\"";
 263
 264                 BOOST_THROW_EXCEPTION(std::runtime_error(mysql_error(&m_Connection)));
 265         }
 266
 267         SetConnected(true);
 268
 269         IdoMysqlResult result = Query("SELECT @@global.max_allowed_packet AS max_allowed_packet");
 270
 271         Dictionary::Ptr row = FetchRow(result);
 272
 273         if (row)
 274                 m_MaxPacketSize = row->Get("max_allowed_packet");
 275         else
 276                 m_MaxPacketSize = 64 * 1024;
 277
 278         DiscardRows(result);
 279
 280         String dbVersionName = "idoutils";
 281         result = Query("SELECT version FROM " + GetTablePrefix() + "dbversion WHERE name='" + Escape(dbVersionName) + "'");
 282
 283         row = FetchRow(result);
 284
 285         if (!row) {
 286                 mysql_close(&m_Connection);
 287                 SetConnected(false);
 288
 289                 Log(LogCritical, "IdoMysqlConnection", "Schema does not provide any valid version! Verify your schema installation.");
 290
 291                 Application::Exit(EXIT_FAILURE);
 292         }
 293
 294         DiscardRows(result);
 295
 296         String version = row->Get("version");
 297
 298         SetSchemaVersion(version);
 299
 300         if (Utility::CompareVersion(IDO_COMPAT_SCHEMA_VERSION, version) < 0) {
 301                 mysql_close(&m_Connection);
 302                 SetConnected(false);
 303
 304                 Log(LogCritical, "IdoMysqlConnection")
 305                     << "Schema version '" << version << "' does not match the required version '"
 306                     << IDO_COMPAT_SCHEMA_VERSION << "' (or newer)! Please check the upgrade documentation at "
 307                     << "https://docs.icinga.com/icinga2/latest/doc/module/icinga2/chapter/upgrading-icinga-2#upgrading-mysql-db";
 308
 309                 Application::Exit(EXIT_FAILURE);
 310         }
 311
 312         String instanceName = GetInstanceName();
 313
 314         result = Query("SELECT instance_id FROM " + GetTablePrefix() + "instances WHERE instance_name = '" + Escape(instanceName) + "'");
 315         row = FetchRow(result);
 316
 317         if (!row) {
 318                 Query("INSERT INTO " + GetTablePrefix() + "instances (instance_name, instance_description) VALUES ('" + Escape(instanceName) + "', '" + Escape(GetInstanceDescription()) + "')");
 319                 m_InstanceID = GetLastInsertID();
 320         } else {
 321                 m_InstanceID = DbReference(row->Get("instance_id"));
 322         }
 323
 324         DiscardRows(result);
 325
 326         Endpoint::Ptr my_endpoint = Endpoint::GetLocalEndpoint();
 327
 328         /* we have an endpoint in a cluster setup, so decide if we can proceed here */
 329         if (my_endpoint && GetHAMode() == HARunOnce) {
 330                 /* get the current endpoint writing to programstatus table */
 331                 result = Query("SELECT UNIX_TIMESTAMP(status_update_time) AS status_update_time, endpoint_name FROM " +
 332                     GetTablePrefix() + "programstatus WHERE instance_id = " + Convert::ToString(m_InstanceID));
 333                 row = FetchRow(result);
 334                 DiscardRows(result);
 335
 336                 String endpoint_name;
 337
 338                 if (row)
 339                         endpoint_name = row->Get("endpoint_name");
 340                 else
 341                         Log(LogNotice, "IdoMysqlConnection", "Empty program status table");
 342
 343                 /* if we did not write into the database earlier, another instance is active */
 344                 if (endpoint_name != my_endpoint->GetName()) {
 345                         double status_update_time;
 346
 347                         if (row)
 348                                 status_update_time = row->Get("status_update_time");
 349                         else
 350                                 status_update_time = 0;
 351
 352                         double status_update_age = Utility::GetTime() - status_update_time;
 353
 354                         Log(LogNotice, "IdoMysqlConnection")
 355                             << "Last update by '" << endpoint_name << "' was " << status_update_age << "s ago.";
 356
 357                         if (status_update_age < GetFailoverTimeout()) {
 358                                 mysql_close(&m_Connection);
 359                                 SetConnected(false);
 360                                 SetShouldConnect(false);
 361
 362                                 return;
 363                         }
 364
 365                         /* activate the IDO only, if we're authoritative in this zone */
 366                         if (IsPaused()) {
 367                                 Log(LogNotice, "IdoMysqlConnection")
 368                                     << "Local endpoint '" << my_endpoint->GetName() << "' is not authoritative, bailing out.";
 369
 370                                 mysql_close(&m_Connection);
 371                                 SetConnected(false);
 372
 373                                 return;
 374                         }
 375                 }
 376
 377                 Log(LogNotice, "IdoMysqlConnection", "Enabling IDO connection.");
 378         }
 379
 380         Log(LogInformation, "IdoMysqlConnection")
 381             << "MySQL IDO instance id: " << static_cast<long>(m_InstanceID) << " (schema version: '" + version + "')";
 382
 383         /* set session time zone to utc */
 384         Query("SET SESSION TIME_ZONE='+00:00'");
 385
 386         Query("SET SESSION SQL_MODE='NO_AUTO_VALUE_ON_ZERO'");
 387
 388         Query("BEGIN");
 389
 390         /* update programstatus table */
 391         UpdateProgramStatus();
 392
 393         /* record connection */
 394         Query("INSERT INTO " + GetTablePrefix() + "conninfo " +
 395             "(instance_id, connect_time, last_checkin_time, agent_name, agent_version, connect_type, data_start_time) VALUES ("
 396             + Convert::ToString(static_cast<long>(m_InstanceID)) + ", NOW(), NOW(), 'icinga2 db_ido_mysql', '" + Escape(Application::GetAppVersion())
 397             + "', '" + (reconnect ? "RECONNECT" : "INITIAL") + "', NOW())");
 398
 399         /* clear config tables for the initial config dump */
 400         PrepareDatabase();
 401
 402         std::ostringstream q1buf;
 403         q1buf << "SELECT object_id, objecttype_id, name1, name2, is_active FROM " + GetTablePrefix() + "objects WHERE instance_id = " << static_cast<long>(m_InstanceID);
 404         result = Query(q1buf.str());
 405
 406         std::vector<DbObject::Ptr> activeDbObjs;
 407
 408         while ((row = FetchRow(result))) {
 409                 DbType::Ptr dbtype = DbType::GetByID(row->Get("objecttype_id"));
 410
 411                 if (!dbtype)
 412                         continue;
 413
 414                 DbObject::Ptr dbobj = dbtype->GetOrCreateObjectByName(row->Get("name1"), row->Get("name2"));
 415                 SetObjectID(dbobj, DbReference(row->Get("object_id")));
 416                 bool active = row->Get("is_active");
 417                 SetObjectActive(dbobj, active);
 418
 419                 if (active)
 420                         activeDbObjs.push_back(dbobj);
 421         }
 422
 423         SetIDCacheValid(true);
 424
 425         EnableActiveChangedHandler();
 426
 427         for (const DbObject::Ptr& dbobj : activeDbObjs) {
 428                 if (dbobj->GetObject())
 429                         continue;
 430
 431                 Log(LogNotice, "IdoMysqlConnection")
 432                     << "Deactivate deleted object name1: '" << dbobj->GetName1()
 433                     << "' name2: '" << dbobj->GetName2() + "'.";
 434                 DeactivateObject(dbobj);
 435         }
 436
 437         UpdateAllObjects();
 438
 439 #ifdef I2_DEBUG /* I2_DEBUG */
 440         Log(LogDebug, "IdoMysqlConnection")
 441             << "Scheduling session table clear and finish connect task.";
 442 #endif /* I2_DEBUG */
 443
 444         m_QueryQueue.Enqueue(boost::bind(&IdoMysqlConnection::ClearTablesBySession, this), PriorityLow);
 445
 446         m_QueryQueue.Enqueue(boost::bind(&IdoMysqlConnection::FinishConnect, this, startTime), PriorityLow);
 447 }
 448
 449 void IdoMysqlConnection::FinishConnect(double startTime)
 450 {
 451         AssertOnWorkQueue();
 452
 453         if (!GetConnected())
 454                 return;
 455
 456         FinishAsyncQueries();
 457
 458         Log(LogInformation, "IdoMysqlConnection")
 459             << "Finished reconnecting to MySQL IDO database in " << std::setw(2) << Utility::GetTime() - startTime << " second(s).";
 460
 461         Query("COMMIT");
 462         Query("BEGIN");
 463 }
 464
 465 void IdoMysqlConnection::ClearTablesBySession(void)
 466 {
 467         /* delete all comments and downtimes without current session token */
 468         ClearTableBySession("comments");
 469         ClearTableBySession("scheduleddowntime");
 470 }
 471
 472 void IdoMysqlConnection::ClearTableBySession(const String& table)
 473 {
 474         Query("DELETE FROM " + GetTablePrefix() + table + " WHERE instance_id = " +
 475             Convert::ToString(static_cast<long>(m_InstanceID)) + " AND session_token <> " +
 476             Convert::ToString(GetSessionToken()));
 477 }
 478
 479 void IdoMysqlConnection::AsyncQuery(const String& query, const boost::function<void (const IdoMysqlResult&)>& callback)
 480 {
 481         AssertOnWorkQueue();
 482
 483         IdoAsyncQuery aq;
 484         aq.Query = query;
 485         /* XXX: Important: The callback must not immediately execute a query, but enqueue it!
 486          * See https://github.com/Icinga/icinga2/issues/4603 for details.
 487          */
 488         aq.Callback = callback;
 489         m_AsyncQueries.push_back(aq);
 490
 491         if (m_AsyncQueries.size() > 25000) {
 492                 FinishAsyncQueries();
 493                 InternalNewTransaction();
 494         }
 495 }
 496
 497 void IdoMysqlConnection::FinishAsyncQueries(void)
 498 {
 499         std::vector<IdoAsyncQuery> queries;
 500         m_AsyncQueries.swap(queries);
 501
 502         std::vector<IdoAsyncQuery>::size_type offset = 0;
 503
 504         while (offset < queries.size()) {
 505                 std::ostringstream querybuf;
 506
 507                 std::vector<IdoAsyncQuery>::size_type count = 0;
 508                 size_t num_bytes = 0;
 509
 510                 for (std::vector<IdoAsyncQuery>::size_type i = offset; i < queries.size(); i++) {
 511                         const IdoAsyncQuery& aq = queries[i];
 512
 513                         size_t size_query = aq.Query.GetLength() + 1;
 514
 515                         if (num_bytes + size_query > m_MaxPacketSize - 512)
 516                                 break;
 517
 518                         if (count > 0)
 519                                 querybuf << ";";
 520
 521                         IncreaseQueryCount();
 522                         count++;
 523
 524                         Log(LogDebug, "IdoMysqlConnection")
 525                             << "Query: " << aq.Query;
 526
 527                         querybuf << aq.Query;
 528                         num_bytes += size_query;
 529                 }
 530
 531                 String query = querybuf.str();
 532
 533                 if (mysql_query(&m_Connection, query.CStr()) != 0) {
 534                         std::ostringstream msgbuf;
 535                         String message = mysql_error(&m_Connection);
 536                         msgbuf << "Error \"" << message << "\" when executing query \"" << query << "\"";
 537                         Log(LogCritical, "IdoMysqlConnection", msgbuf.str());
 538
 539                         BOOST_THROW_EXCEPTION(
 540                             database_error()
 541                                 << errinfo_message(mysql_error(&m_Connection))
 542                                 << errinfo_database_query(query)
 543                         );
 544                 }
 545
 546                 for (std::vector<IdoAsyncQuery>::size_type i = offset; i < offset + count; i++) {
 547                         const IdoAsyncQuery& aq = queries[i];
 548
 549                         MYSQL_RES *result = mysql_store_result(&m_Connection);
 550
 551                         m_AffectedRows = mysql_affected_rows(&m_Connection);
 552
 553                         IdoMysqlResult iresult;
 554
 555                         if (!result) {
 556                                 if (mysql_field_count(&m_Connection) > 0) {
 557                                         std::ostringstream msgbuf;
 558                                         String message = mysql_error(&m_Connection);
 559                                         msgbuf << "Error \"" << message << "\" when executing query \"" << aq.Query << "\"";
 560                                         Log(LogCritical, "IdoMysqlConnection", msgbuf.str());
 561
 562                                         BOOST_THROW_EXCEPTION(
 563                                             database_error()
 564                                                 << errinfo_message(mysql_error(&m_Connection))
 565                                                 << errinfo_database_query(query)
 566                                         );
 567                                 }
 568                         } else
 569                                 iresult = IdoMysqlResult(result, std::ptr_fun(mysql_free_result));
 570
 571                         if (aq.Callback)
 572                                 aq.Callback(iresult);
 573
 574                         if (mysql_next_result(&m_Connection) > 0) {
 575                                 std::ostringstream msgbuf;
 576                                 String message = mysql_error(&m_Connection);
 577                                 msgbuf << "Error \"" << message << "\" when executing query \"" << query << "\"";
 578                                 Log(LogCritical, "IdoMysqlConnection", msgbuf.str());
 579
 580                                 BOOST_THROW_EXCEPTION(
 581                                     database_error()
 582                                         << errinfo_message(mysql_error(&m_Connection))
 583                                         << errinfo_database_query(query)
 584                                 );
 585                         }
 586                 }
 587
 588                 offset += count;
 589         }
 590 }
 591
 592 IdoMysqlResult IdoMysqlConnection::Query(const String& query)
 593 {
 594         AssertOnWorkQueue();
 595
 596         /* finish all async queries to maintain the right order for queries */
 597         FinishAsyncQueries();
 598
 599         Log(LogDebug, "IdoMysqlConnection")
 600             << "Query: " << query;
 601
 602         IncreaseQueryCount();
 603
 604         if (mysql_query(&m_Connection, query.CStr()) != 0) {
 605                 std::ostringstream msgbuf;
 606                 String message = mysql_error(&m_Connection);
 607                 msgbuf << "Error \"" << message << "\" when executing query \"" << query << "\"";
 608                 Log(LogCritical, "IdoMysqlConnection", msgbuf.str());
 609
 610                 BOOST_THROW_EXCEPTION(
 611                     database_error()
 612                         << errinfo_message(mysql_error(&m_Connection))
 613                         << errinfo_database_query(query)
 614                 );
 615         }
 616
 617         MYSQL_RES *result = mysql_store_result(&m_Connection);
 618
 619         m_AffectedRows = mysql_affected_rows(&m_Connection);
 620
 621         if (!result) {
 622                 if (mysql_field_count(&m_Connection) > 0) {
 623                         std::ostringstream msgbuf;
 624                         String message = mysql_error(&m_Connection);
 625                         msgbuf << "Error \"" << message << "\" when executing query \"" << query << "\"";
 626                         Log(LogCritical, "IdoMysqlConnection", msgbuf.str());
 627
 628                         BOOST_THROW_EXCEPTION(
 629                             database_error()
 630                                 << errinfo_message(mysql_error(&m_Connection))
 631                                 << errinfo_database_query(query)
 632                         );
 633                 }
 634
 635                 return IdoMysqlResult();
 636         }
 637
 638         return IdoMysqlResult(result, std::ptr_fun(mysql_free_result));
 639 }
 640
 641 DbReference IdoMysqlConnection::GetLastInsertID(void)
 642 {
 643         AssertOnWorkQueue();
 644
 645         return DbReference(mysql_insert_id(&m_Connection));
 646 }
 647
 648 int IdoMysqlConnection::GetAffectedRows(void)
 649 {
 650         AssertOnWorkQueue();
 651
 652         return m_AffectedRows;
 653 }
 654
 655 String IdoMysqlConnection::Escape(const String& s)
 656 {
 657         AssertOnWorkQueue();
 658
 659         String utf8s = Utility::ValidateUTF8(s);
 660
 661         size_t length = utf8s.GetLength();
 662         char *to = new char[utf8s.GetLength() * 2 + 1];
 663
 664         mysql_real_escape_string(&m_Connection, to, utf8s.CStr(), length);
 665
 666         String result = String(to);
 667
 668         delete [] to;
 669
 670         return result;
 671 }
 672
 673 Dictionary::Ptr IdoMysqlConnection::FetchRow(const IdoMysqlResult& result)
 674 {
 675         AssertOnWorkQueue();
 676
 677         MYSQL_ROW row;
 678         MYSQL_FIELD *field;
 679         unsigned long *lengths, i;
 680
 681         row = mysql_fetch_row(result.get());
 682
 683         if (!row)
 684                 return Dictionary::Ptr();
 685
 686         lengths = mysql_fetch_lengths(result.get());
 687
 688         if (!lengths)
 689                 return Dictionary::Ptr();
 690
 691         Dictionary::Ptr dict = new Dictionary();
 692
 693         mysql_field_seek(result.get(), 0);
 694         for (field = mysql_fetch_field(result.get()), i = 0; field; field = mysql_fetch_field(result.get()), i++)
 695                 dict->Set(field->name, String(row[i], row[i] + lengths[i]));
 696
 697         return dict;
 698 }
 699
 700 void IdoMysqlConnection::DiscardRows(const IdoMysqlResult& result)
 701 {
 702         Dictionary::Ptr row;
 703
 704         while ((row = FetchRow(result)))
 705                 ; /* empty loop body */
 706 }
 707
 708 void IdoMysqlConnection::ActivateObject(const DbObject::Ptr& dbobj)
 709 {
 710 #ifdef I2_DEBUG /* I2_DEBUG */
 711         Log(LogDebug, "IdoMysqlConnection")
 712             << "Scheduling object activation task for '" << dbobj->GetName1() << "!" << dbobj->GetName2() << "'.";
 713 #endif /* I2_DEBUG */
 714
 715         m_QueryQueue.Enqueue(boost::bind(&IdoMysqlConnection::InternalActivateObject, this, dbobj), PriorityLow);
 716 }
 717
 718 void IdoMysqlConnection::InternalActivateObject(const DbObject::Ptr& dbobj)
 719 {
 720         AssertOnWorkQueue();
 721
 722         if (!GetConnected())
 723                 return;
 724
 725         DbReference dbref = GetObjectID(dbobj);
 726         std::ostringstream qbuf;
 727
 728         if (!dbref.IsValid()) {
 729                 if (!dbobj->GetName2().IsEmpty()) {
 730                         qbuf << "INSERT INTO " + GetTablePrefix() + "objects (instance_id, objecttype_id, name1, name2, is_active) VALUES ("
 731                              << static_cast<long>(m_InstanceID) << ", " << dbobj->GetType()->GetTypeID() << ", "
 732                              << "'" << Escape(dbobj->GetName1()) << "', '" << Escape(dbobj->GetName2()) << "', 1)";
 733                 } else {
 734                         qbuf << "INSERT INTO " + GetTablePrefix() + "objects (instance_id, objecttype_id, name1, is_active) VALUES ("
 735                              << static_cast<long>(m_InstanceID) << ", " << dbobj->GetType()->GetTypeID() << ", "
 736                              << "'" << Escape(dbobj->GetName1()) << "', 1)";
 737                 }
 738
 739                 Query(qbuf.str());
 740                 SetObjectID(dbobj, GetLastInsertID());
 741         } else {
 742                 qbuf << "UPDATE " + GetTablePrefix() + "objects SET is_active = 1 WHERE object_id = " << static_cast<long>(dbref);
 743                 AsyncQuery(qbuf.str());
 744         }
 745 }
 746
 747 void IdoMysqlConnection::DeactivateObject(const DbObject::Ptr& dbobj)
 748 {
 749 #ifdef I2_DEBUG /* I2_DEBUG */
 750         Log(LogDebug, "IdoMysqlConnection")
 751             << "Scheduling object deactivation task for '" << dbobj->GetName1() << "!" << dbobj->GetName2() << "'.";
 752 #endif /* I2_DEBUG */
 753
 754         m_QueryQueue.Enqueue(boost::bind(&IdoMysqlConnection::InternalDeactivateObject, this, dbobj), PriorityLow);
 755 }
 756
 757 void IdoMysqlConnection::InternalDeactivateObject(const DbObject::Ptr& dbobj)
 758 {
 759         AssertOnWorkQueue();
 760
 761         if (!GetConnected())
 762                 return;
 763
 764         DbReference dbref = GetObjectID(dbobj);
 765
 766         if (!dbref.IsValid())
 767                 return;
 768
 769         std::ostringstream qbuf;
 770         qbuf << "UPDATE " + GetTablePrefix() + "objects SET is_active = 0 WHERE object_id = " << static_cast<long>(dbref);
 771         AsyncQuery(qbuf.str());
 772
 773         /* Note that we're _NOT_ clearing the db refs via SetReference/SetConfigUpdate/SetStatusUpdate
 774          * because the object is still in the database. */
 775 }
 776
 777 bool IdoMysqlConnection::FieldToEscapedString(const String& key, const Value& value, Value *result)
 778 {
 779         if (key == "instance_id") {
 780                 *result = static_cast<long>(m_InstanceID);
 781                 return true;
 782         } else if (key == "session_token") {
 783                 *result = GetSessionToken();
 784                 return true;
 785         }
 786
 787         Value rawvalue = DbValue::ExtractValue(value);
 788
 789         if (rawvalue.IsObjectType<ConfigObject>()) {
 790                 DbObject::Ptr dbobjcol = DbObject::GetOrCreateByObject(rawvalue);
 791
 792                 if (!dbobjcol) {
 793                         *result = 0;
 794                         return true;
 795                 }
 796
 797                 if (!IsIDCacheValid())
 798                         return false;
 799
 800                 DbReference dbrefcol;
 801
 802                 if (DbValue::IsObjectInsertID(value)) {
 803                         dbrefcol = GetInsertID(dbobjcol);
 804
 805                         if (!dbrefcol.IsValid())
 806                                 return false;
 807                 } else {
 808                         dbrefcol = GetObjectID(dbobjcol);
 809
 810                         if (!dbrefcol.IsValid()) {
 811                                 InternalActivateObject(dbobjcol);
 812
 813                                 dbrefcol = GetObjectID(dbobjcol);
 814
 815                                 if (!dbrefcol.IsValid())
 816                                         return false;
 817                         }
 818                 }
 819
 820                 *result = static_cast<long>(dbrefcol);
 821         } else if (DbValue::IsTimestamp(value)) {
 822                 long ts = rawvalue;
 823                 std::ostringstream msgbuf;
 824                 msgbuf << "FROM_UNIXTIME(" << ts << ")";
 825                 *result = Value(msgbuf.str());
 826         } else if (DbValue::IsTimestampNow(value)) {
 827                 *result = "NOW()";
 828         } else if (DbValue::IsObjectInsertID(value)) {
 829                 long id = static_cast<long>(rawvalue);
 830
 831                 if (id <= 0)
 832                         return false;
 833
 834                 *result = id;
 835                 return true;
 836         } else {
 837                 Value fvalue;
 838
 839                 if (rawvalue.IsBoolean())
 840                         fvalue = Convert::ToLong(rawvalue);
 841                 else
 842                         fvalue = rawvalue;
 843
 844                 *result = "'" + Escape(fvalue) + "'";
 845         }
 846
 847         return true;
 848 }
 849
 850 void IdoMysqlConnection::ExecuteQuery(const DbQuery& query)
 851 {
 852         ASSERT(query.Category != DbCatInvalid);
 853
 854 #ifdef I2_DEBUG /* I2_DEBUG */
 855         Log(LogDebug, "IdoMysqlConnection")
 856             << "Scheduling execute query task, type " << query.Type << ", table '" << query.Table << "'.";
 857 #endif /* I2_DEBUG */
 858
 859         m_QueryQueue.Enqueue(boost::bind(&IdoMysqlConnection::InternalExecuteQuery, this, query, -1), query.Priority, true);
 860 }
 861
 862 void IdoMysqlConnection::ExecuteMultipleQueries(const std::vector<DbQuery>& queries)
 863 {
 864         if (queries.empty())
 865                 return;
 866
 867 #ifdef I2_DEBUG /* I2_DEBUG */
 868         Log(LogDebug, "IdoMysqlConnection")
 869             << "Scheduling multiple execute query task, type " << queries[0].Type << ", table '" << queries[0].Table << "'.";
 870 #endif /* I2_DEBUG */
 871
 872         m_QueryQueue.Enqueue(boost::bind(&IdoMysqlConnection::InternalExecuteMultipleQueries, this, queries), queries[0].Priority, true);
 873 }
 874
 875 bool IdoMysqlConnection::CanExecuteQuery(const DbQuery& query)
 876 {
 877         if (query.Object && !IsIDCacheValid())
 878                 return false;
 879
 880         if (query.WhereCriteria) {
 881                 ObjectLock olock(query.WhereCriteria);
 882                 Value value;
 883
 884                 for (const Dictionary::Pair& kv : query.WhereCriteria) {
 885                         if (!FieldToEscapedString(kv.first, kv.second, &value))
 886                                 return false;
 887                 }
 888         }
 889
 890         if (query.Fields) {
 891                 ObjectLock olock(query.Fields);
 892
 893                 for (const Dictionary::Pair& kv : query.Fields) {
 894                         Value value;
 895
 896                         if (kv.second.IsEmpty() && !kv.second.IsString())
 897                                 continue;
 898
 899                         if (!FieldToEscapedString(kv.first, kv.second, &value))
 900                                 return false;
 901                 }
 902         }
 903
 904         return true;
 905 }
 906
 907 void IdoMysqlConnection::InternalExecuteMultipleQueries(const std::vector<DbQuery>& queries)
 908 {
 909         AssertOnWorkQueue();
 910
 911         if (!GetConnected())
 912                 return;
 913
 914         for (const DbQuery& query : queries) {
 915                 ASSERT(query.Type == DbQueryNewTransaction || query.Category != DbCatInvalid);
 916
 917                 if (!CanExecuteQuery(query)) {
 918
 919 #ifdef I2_DEBUG /* I2_DEBUG */
 920                         Log(LogDebug, "IdoMysqlConnection")
 921                             << "Scheduling multiple execute query task again: Cannot execute query now. Type '"
 922                             << query.Type << "', table '" << query.Table << "', queue size: '" << GetPendingQueryCount() << "'.";
 923 #endif /* I2_DEBUG */
 924
 925                         m_QueryQueue.Enqueue(boost::bind(&IdoMysqlConnection::InternalExecuteMultipleQueries, this, queries), query.Priority);
 926                         return;
 927                 }
 928         }
 929
 930         for (const DbQuery& query : queries) {
 931                 InternalExecuteQuery(query);
 932         }
 933 }
 934
 935 void IdoMysqlConnection::InternalExecuteQuery(const DbQuery& query, int typeOverride)
 936 {
 937         AssertOnWorkQueue();
 938
 939         if (!GetConnected())
 940                 return;
 941
 942         if (query.Type == DbQueryNewTransaction) {
 943                 InternalNewTransaction();
 944                 return;
 945         }
 946
 947         /* check whether we're allowed to execute the query first */
 948         if (GetCategoryFilter() != DbCatEverything && (query.Category & GetCategoryFilter()) == 0)
 949                 return;
 950
 951         if (query.Object && query.Object->GetObject()->GetExtension("agent_check").ToBool())
 952                 return;
 953
 954         /* check if there are missing object/insert ids and re-enqueue the query */
 955         if (!CanExecuteQuery(query)) {
 956
 957 #ifdef I2_DEBUG /* I2_DEBUG */
 958                 Log(LogDebug, "IdoMysqlConnection")
 959                     << "Scheduling execute query task again: Cannot execute query now. Type '"
 960                     << typeOverride << "', table '" << query.Table << "', queue size: '" << GetPendingQueryCount() << "'.";
 961 #endif /* I2_DEBUG */
 962
 963                 m_QueryQueue.Enqueue(boost::bind(&IdoMysqlConnection::InternalExecuteQuery, this, query, typeOverride), query.Priority);
 964                 return;
 965         }
 966
 967         std::ostringstream qbuf, where;
 968         int type;
 969
 970         if (query.WhereCriteria) {
 971                 where << " WHERE ";
 972
 973                 ObjectLock olock(query.WhereCriteria);
 974                 Value value;
 975                 bool first = true;
 976
 977                 for (const Dictionary::Pair& kv : query.WhereCriteria) {
 978                         if (!FieldToEscapedString(kv.first, kv.second, &value)) {
 979
 980 #ifdef I2_DEBUG /* I2_DEBUG */
 981                                 Log(LogDebug, "IdoMysqlConnection")
 982                                     << "Scheduling execute query task again: Cannot execute query now. Type '"
 983                                     << typeOverride << "', table '" << query.Table << "', queue size: '" << GetPendingQueryCount() << "'.";
 984 #endif /* I2_DEBUG */
 985
 986                                 m_QueryQueue.Enqueue(boost::bind(&IdoMysqlConnection::InternalExecuteQuery, this, query, -1), query.Priority);
 987                                 return;
 988                         }
 989
 990                         if (!first)
 991                                 where << " AND ";
 992
 993                         where << kv.first << " = " << value;
 994
 995                         if (first)
 996                                 first = false;
 997                 }
 998         }
 999
1000         type = (typeOverride != -1) ? typeOverride : query.Type;
1001
1002         bool upsert = false;
1003
1004         if ((type & DbQueryInsert) && (type & DbQueryUpdate)) {
1005                 bool hasid = false;
1006
1007                 if (query.Object) {
1008                         if (query.ConfigUpdate)
1009                                 hasid = GetConfigUpdate(query.Object);
1010                         else if (query.StatusUpdate)
1011                                 hasid = GetStatusUpdate(query.Object);
1012                 }
1013
1014                 if (!hasid)
1015                         upsert = true;
1016
1017                 type = DbQueryUpdate;
1018         }
1019
1020         if ((type & DbQueryInsert) && (type & DbQueryDelete)) {
1021                 std::ostringstream qdel;
1022                 qdel << "DELETE FROM " << GetTablePrefix() << query.Table << where.str();
1023                 AsyncQuery(qdel.str());
1024
1025                 type = DbQueryInsert;
1026         }
1027
1028         switch (type) {
1029                 case DbQueryInsert:
1030                         qbuf << "INSERT INTO " << GetTablePrefix() << query.Table;
1031                         break;
1032                 case DbQueryUpdate:
1033                         qbuf << "UPDATE " << GetTablePrefix() << query.Table << " SET";
1034                         break;
1035                 case DbQueryDelete:
1036                         qbuf << "DELETE FROM " << GetTablePrefix() << query.Table;
1037                         break;
1038                 default:
1039                         VERIFY(!"Invalid query type.");
1040         }
1041
1042         if (type == DbQueryInsert || type == DbQueryUpdate) {
1043                 std::ostringstream colbuf, valbuf;
1044
1045                 if (type == DbQueryUpdate && query.Fields->GetLength() == 0)
1046                         return;
1047
1048                 ObjectLock olock(query.Fields);
1049
1050                 bool first = true;
1051                 for (const Dictionary::Pair& kv : query.Fields) {
1052                         Value value;
1053
1054                         if (kv.second.IsEmpty() && !kv.second.IsString())
1055                                 continue;
1056
1057                         if (!FieldToEscapedString(kv.first, kv.second, &value)) {
1058
1059 #ifdef I2_DEBUG /* I2_DEBUG */
1060                                 Log(LogDebug, "IdoMysqlConnection")
1061                                     << "Scheduling execute query task again: Cannot extract required INSERT/UPDATE fields, key '"
1062                                     << kv.first << "', val '" << kv.second << "', type " << typeOverride << ", table '" << query.Table << "'.";
1063 #endif /* I2_DEBUG */
1064
1065                                 m_QueryQueue.Enqueue(boost::bind(&IdoMysqlConnection::InternalExecuteQuery, this, query, -1), query.Priority);
1066                                 return;
1067                         }
1068
1069                         if (type == DbQueryInsert) {
1070                                 if (!first) {
1071                                         colbuf << ", ";
1072                                         valbuf << ", ";
1073                                 }
1074
1075                                 colbuf << kv.first;
1076                                 valbuf << value;
1077                         } else {
1078                                 if (!first)
1079                                         qbuf << ", ";
1080
1081                                 qbuf << " " << kv.first << " = " << value;
1082                         }
1083
1084                         if (first)
1085                                 first = false;
1086                 }
1087
1088                 if (type == DbQueryInsert)
1089                         qbuf << " (" << colbuf.str() << ") VALUES (" << valbuf.str() << ")";
1090         }
1091
1092         if (type != DbQueryInsert)
1093                 qbuf << where.str();
1094
1095         AsyncQuery(qbuf.str(), boost::bind(&IdoMysqlConnection::FinishExecuteQuery, this, query, type, upsert));
1096 }
1097
1098 void IdoMysqlConnection::FinishExecuteQuery(const DbQuery& query, int type, bool upsert)
1099 {
1100         if (upsert && GetAffectedRows() == 0) {
1101
1102 #ifdef I2_DEBUG /* I2_DEBUG */
1103                 Log(LogDebug, "IdoMysqlConnection")
1104                     << "Rescheduling DELETE/INSERT query: Upsert UPDATE did not affect rows, type " << type << ", table '" << query.Table << "'.";
1105 #endif /* I2_DEBUG */
1106
1107                 m_QueryQueue.Enqueue(boost::bind(&IdoMysqlConnection::InternalExecuteQuery, this, query, DbQueryDelete | DbQueryInsert), query.Priority);
1108
1109                 return;
1110         }
1111
1112         if (type == DbQueryInsert && query.Object) {
1113                 if (query.ConfigUpdate) {
1114                         SetInsertID(query.Object, GetLastInsertID());
1115                         SetConfigUpdate(query.Object, true);
1116                 } else if (query.StatusUpdate)
1117                         SetStatusUpdate(query.Object, true);
1118         }
1119
1120         if (type == DbQueryInsert && query.Table == "notifications" && query.NotificationInsertID)
1121                 query.NotificationInsertID->SetValue(static_cast<long>(GetLastInsertID()));
1122 }
1123
1124 void IdoMysqlConnection::CleanUpExecuteQuery(const String& table, const String& time_column, double max_age)
1125 {
1126 #ifdef I2_DEBUG /* I2_DEBUG */
1127                 Log(LogDebug, "IdoMysqlConnection")
1128                     << "Rescheduling cleanup query for table '" << table << "' and column '"
1129                     << time_column << "'. max_age is set to '" << max_age << "'.";
1130 #endif /* I2_DEBUG */
1131
1132         m_QueryQueue.Enqueue(boost::bind(&IdoMysqlConnection::InternalCleanUpExecuteQuery, this, table, time_column, max_age), PriorityLow, true);
1133 }
1134
1135 void IdoMysqlConnection::InternalCleanUpExecuteQuery(const String& table, const String& time_column, double max_age)
1136 {
1137         AssertOnWorkQueue();
1138
1139         if (!GetConnected())
1140                 return;
1141
1142         AsyncQuery("DELETE FROM " + GetTablePrefix() + table + " WHERE instance_id = " +
1143             Convert::ToString(static_cast<long>(m_InstanceID)) + " AND " + time_column +
1144             " < FROM_UNIXTIME(" + Convert::ToString(static_cast<long>(max_age)) + ")");
1145 }
1146
1147 void IdoMysqlConnection::FillIDCache(const DbType::Ptr& type)
1148 {
1149         String query = "SELECT " + type->GetIDColumn() + " AS object_id, " + type->GetTable() + "_id, config_hash FROM " + GetTablePrefix() + type->GetTable() + "s";
1150         IdoMysqlResult result = Query(query);
1151
1152         Dictionary::Ptr row;
1153
1154         while ((row = FetchRow(result))) {
1155                 DbReference dbref(row->Get("object_id"));
1156                 SetInsertID(type, dbref, DbReference(row->Get(type->GetTable() + "_id")));
1157                 SetConfigHash(type, dbref, row->Get("config_hash"));
1158         }
1159 }
1160
1161 int IdoMysqlConnection::GetPendingQueryCount(void) const
1162 {
1163         return m_QueryQueue.GetLength();
1164 }