granicus.if.org Git - icinga2/blob - lib/db_ido_pgsql/idopgsqlconnection.cpp

   1 /******************************************************************************
   2  * Icinga 2                                                                   *
   3  * Copyright (C) 2012-2018 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_pgsql/idopgsqlconnection.hpp"
  21 #include "db_ido_pgsql/idopgsqlconnection.tcpp"
  22 #include "db_ido/dbtype.hpp"
  23 #include "db_ido/dbvalue.hpp"
  24 #include "base/logger.hpp"
  25 #include "base/objectlock.hpp"
  26 #include "base/convert.hpp"
  27 #include "base/utility.hpp"
  28 #include "base/perfdatavalue.hpp"
  29 #include "base/application.hpp"
  30 #include "base/configtype.hpp"
  31 #include "base/exception.hpp"
  32 #include "base/context.hpp"
  33 #include "base/statsfunction.hpp"
  34 #include <boost/tuple/tuple.hpp>
  35 #include <utility>
  36
  37 using namespace icinga;
  38
  39 REGISTER_TYPE(IdoPgsqlConnection);
  40
  41 REGISTER_STATSFUNCTION(IdoPgsqlConnection, &IdoPgsqlConnection::StatsFunc);
  42
  43 IdoPgsqlConnection::IdoPgsqlConnection()
  44 {
  45         m_QueryQueue.SetName("IdoPgsqlConnection, " + GetName());
  46 }
  47
  48 void IdoPgsqlConnection::OnConfigLoaded()
  49 {
  50         ObjectImpl<IdoPgsqlConnection>::OnConfigLoaded();
  51
  52         m_QueryQueue.SetName("IdoPgsqlConnection, " + GetName());
  53
  54         Library shimLibrary{"pgsql_shim"};
  55
  56         auto create_pgsql_shim = shimLibrary.GetSymbolAddress<create_pgsql_shim_ptr>("create_pgsql_shim");
  57
  58         m_Pgsql.reset(create_pgsql_shim());
  59
  60         std::swap(m_Library, shimLibrary);
  61 }
  62
  63 void IdoPgsqlConnection::StatsFunc(const Dictionary::Ptr& status, const Array::Ptr& perfdata)
  64 {
  65         Dictionary::Ptr nodes = new Dictionary();
  66
  67         for (const IdoPgsqlConnection::Ptr& idopgsqlconnection : ConfigType::GetObjectsByType<IdoPgsqlConnection>()) {
  68                 size_t queryQueueItems = idopgsqlconnection->m_QueryQueue.GetLength();
  69                 double queryQueueItemRate = idopgsqlconnection->m_QueryQueue.GetTaskCount(60) / 60.0;
  70
  71                 Dictionary::Ptr stats = new Dictionary();
  72                 stats->Set("version", idopgsqlconnection->GetSchemaVersion());
  73                 stats->Set("instance_name", idopgsqlconnection->GetInstanceName());
  74                 stats->Set("connected", idopgsqlconnection->GetConnected());
  75                 stats->Set("query_queue_items", queryQueueItems);
  76                 stats->Set("query_queue_item_rate", queryQueueItemRate);
  77
  78                 nodes->Set(idopgsqlconnection->GetName(), stats);
  79
  80                 perfdata->Add(new PerfdataValue("idopgsqlconnection_" + idopgsqlconnection->GetName() + "_queries_rate", idopgsqlconnection->GetQueryCount(60) / 60.0));
  81                 perfdata->Add(new PerfdataValue("idopgsqlconnection_" + idopgsqlconnection->GetName() + "_queries_1min", idopgsqlconnection->GetQueryCount(60)));
  82                 perfdata->Add(new PerfdataValue("idopgsqlconnection_" + idopgsqlconnection->GetName() + "_queries_5mins", idopgsqlconnection->GetQueryCount(5 * 60)));
  83                 perfdata->Add(new PerfdataValue("idopgsqlconnection_" + idopgsqlconnection->GetName() + "_queries_15mins", idopgsqlconnection->GetQueryCount(15 * 60)));
  84                 perfdata->Add(new PerfdataValue("idopgsqlconnection_" + idopgsqlconnection->GetName() + "_query_queue_items", queryQueueItems));
  85                 perfdata->Add(new PerfdataValue("idopgsqlconnection_" + idopgsqlconnection->GetName() + "_query_queue_item_rate", queryQueueItemRate));
  86         }
  87
  88         status->Set("idopgsqlconnection", nodes);
  89 }
  90
  91 void IdoPgsqlConnection::Resume()
  92 {
  93         DbConnection::Resume();
  94
  95         Log(LogInformation, "IdoPgsqlConnection")
  96                 << "'" << GetName() << "' resumed.";
  97
  98         SetConnected(false);
  99
 100         m_QueryQueue.SetExceptionCallback(std::bind(&IdoPgsqlConnection::ExceptionHandler, this, _1));
 101
 102         m_TxTimer = new Timer();
 103         m_TxTimer->SetInterval(1);
 104         m_TxTimer->OnTimerExpired.connect(std::bind(&IdoPgsqlConnection::TxTimerHandler, this));
 105         m_TxTimer->Start();
 106
 107         m_ReconnectTimer = new Timer();
 108         m_ReconnectTimer->SetInterval(10);
 109         m_ReconnectTimer->OnTimerExpired.connect(std::bind(&IdoPgsqlConnection::ReconnectTimerHandler, this));
 110         m_ReconnectTimer->Start();
 111         m_ReconnectTimer->Reschedule(0);
 112
 113         ASSERT(m_Pgsql->isthreadsafe());
 114 }
 115
 116 void IdoPgsqlConnection::Pause()
 117 {
 118         Log(LogInformation, "IdoPgsqlConnection")
 119                 << "'" << GetName() << "' paused.";
 120
 121         m_ReconnectTimer.reset();
 122
 123         DbConnection::Pause();
 124
 125         m_QueryQueue.Enqueue(std::bind(&IdoPgsqlConnection::Disconnect, this), PriorityHigh);
 126         m_QueryQueue.Join();
 127 }
 128
 129 void IdoPgsqlConnection::ExceptionHandler(boost::exception_ptr exp)
 130 {
 131         Log(LogWarning, "IdoPgsqlConnection", "Exception during database operation: Verify that your database is operational!");
 132
 133         Log(LogDebug, "IdoPgsqlConnection")
 134                 << "Exception during database operation: " << DiagnosticInformation(std::move(exp));
 135
 136         if (GetConnected()) {
 137                 m_Pgsql->finish(m_Connection);
 138                 SetConnected(false);
 139         }
 140 }
 141
 142 void IdoPgsqlConnection::AssertOnWorkQueue()
 143 {
 144         ASSERT(m_QueryQueue.IsWorkerThread());
 145 }
 146
 147 void IdoPgsqlConnection::Disconnect()
 148 {
 149         AssertOnWorkQueue();
 150
 151         if (!GetConnected())
 152                 return;
 153
 154         Query("COMMIT");
 155
 156         m_Pgsql->finish(m_Connection);
 157         SetConnected(false);
 158 }
 159
 160 void IdoPgsqlConnection::TxTimerHandler()
 161 {
 162         NewTransaction();
 163 }
 164
 165 void IdoPgsqlConnection::NewTransaction()
 166 {
 167         m_QueryQueue.Enqueue(std::bind(&IdoPgsqlConnection::InternalNewTransaction, this), PriorityHigh, true);
 168 }
 169
 170 void IdoPgsqlConnection::InternalNewTransaction()
 171 {
 172         AssertOnWorkQueue();
 173
 174         if (!GetConnected())
 175                 return;
 176
 177         Query("COMMIT");
 178         Query("BEGIN");
 179 }
 180
 181 void IdoPgsqlConnection::ReconnectTimerHandler()
 182 {
 183         m_QueryQueue.Enqueue(std::bind(&IdoPgsqlConnection::Reconnect, this), PriorityLow);
 184 }
 185
 186 void IdoPgsqlConnection::Reconnect()
 187 {
 188         AssertOnWorkQueue();
 189
 190         CONTEXT("Reconnecting to PostgreSQL IDO database '" + GetName() + "'");
 191
 192         double startTime = Utility::GetTime();
 193
 194         SetShouldConnect(true);
 195
 196         bool reconnect = false;
 197
 198         if (GetConnected()) {
 199                 /* Check if we're really still connected */
 200                 try {
 201                         Query("SELECT 1");
 202                         return;
 203                 } catch (const std::exception&) {
 204                         m_Pgsql->finish(m_Connection);
 205                         SetConnected(false);
 206                         reconnect = true;
 207                 }
 208         }
 209
 210         ClearIDCache();
 211
 212         String ihost, iport, iuser, ipasswd, idb;
 213         const char *host, *port, *user , *passwd, *db;
 214
 215         ihost = GetHost();
 216         iport = GetPort();
 217         iuser = GetUser();
 218         ipasswd = GetPassword();
 219         idb = GetDatabase();
 220
 221         host = (!ihost.IsEmpty()) ? ihost.CStr() : nullptr;
 222         port = (!iport.IsEmpty()) ? iport.CStr() : nullptr;
 223         user = (!iuser.IsEmpty()) ? iuser.CStr() : nullptr;
 224         passwd = (!ipasswd.IsEmpty()) ? ipasswd.CStr() : nullptr;
 225         db = (!idb.IsEmpty()) ? idb.CStr() : nullptr;
 226
 227         m_Connection = m_Pgsql->setdbLogin(host, port, nullptr, nullptr, db, user, passwd);
 228
 229         if (!m_Connection)
 230                 return;
 231
 232         if (m_Pgsql->status(m_Connection) != CONNECTION_OK) {
 233                 String message = m_Pgsql->errorMessage(m_Connection);
 234                 m_Pgsql->finish(m_Connection);
 235                 SetConnected(false);
 236
 237                 Log(LogCritical, "IdoPgsqlConnection")
 238                         << "Connection to database '" << db << "' with user '" << user << "' on '" << host << ":" << port
 239                         << "' failed: \"" << message << "\"";
 240
 241                 BOOST_THROW_EXCEPTION(std::runtime_error(message));
 242         }
 243
 244         SetConnected(true);
 245
 246         IdoPgsqlResult result;
 247
 248         /* explicitely require legacy mode for string escaping in PostgreSQL >= 9.1
 249          * changing standard_conforming_strings to on by default
 250          */
 251         if (m_Pgsql->serverVersion(m_Connection) >= 90100)
 252                 result = Query("SET standard_conforming_strings TO off");
 253
 254         String dbVersionName = "idoutils";
 255         result = Query("SELECT version FROM " + GetTablePrefix() + "dbversion WHERE name=E'" + Escape(dbVersionName) + "'");
 256
 257         Dictionary::Ptr row = FetchRow(result, 0);
 258
 259         if (!row) {
 260                 m_Pgsql->finish(m_Connection);
 261                 SetConnected(false);
 262
 263                 Log(LogCritical, "IdoPgsqlConnection", "Schema does not provide any valid version! Verify your schema installation.");
 264
 265                 BOOST_THROW_EXCEPTION(std::runtime_error("Invalid schema."));
 266         }
 267
 268         String version = row->Get("version");
 269
 270         SetSchemaVersion(version);
 271
 272         if (Utility::CompareVersion(IDO_COMPAT_SCHEMA_VERSION, version) < 0) {
 273                 m_Pgsql->finish(m_Connection);
 274                 SetConnected(false);
 275
 276                 Log(LogCritical, "IdoPgsqlConnection")
 277                         << "Schema version '" << version << "' does not match the required version '"
 278                         << IDO_COMPAT_SCHEMA_VERSION << "' (or newer)! Please check the upgrade documentation at "
 279                         << "https://docs.icinga.com/icinga2/latest/doc/module/icinga2/chapter/upgrading-icinga-2#upgrading-postgresql-db";
 280
 281                 BOOST_THROW_EXCEPTION(std::runtime_error("Schema version mismatch."));
 282         }
 283
 284         String instanceName = GetInstanceName();
 285
 286         result = Query("SELECT instance_id FROM " + GetTablePrefix() + "instances WHERE instance_name = E'" + Escape(instanceName) + "'");
 287         row = FetchRow(result, 0);
 288
 289         if (!row) {
 290                 Query("INSERT INTO " + GetTablePrefix() + "instances (instance_name, instance_description) VALUES (E'" + Escape(instanceName) + "', E'" + Escape(GetInstanceDescription()) + "')");
 291                 m_InstanceID = GetSequenceValue(GetTablePrefix() + "instances", "instance_id");
 292         } else {
 293                 m_InstanceID = DbReference(row->Get("instance_id"));
 294         }
 295
 296         Endpoint::Ptr my_endpoint = Endpoint::GetLocalEndpoint();
 297
 298         /* we have an endpoint in a cluster setup, so decide if we can proceed here */
 299         if (my_endpoint && GetHAMode() == HARunOnce) {
 300                 /* get the current endpoint writing to programstatus table */
 301                 result = Query("SELECT UNIX_TIMESTAMP(status_update_time) AS status_update_time, endpoint_name FROM " +
 302                         GetTablePrefix() + "programstatus WHERE instance_id = " + Convert::ToString(m_InstanceID));
 303                 row = FetchRow(result, 0);
 304
 305                 String endpoint_name;
 306
 307                 if (row)
 308                         endpoint_name = row->Get("endpoint_name");
 309                 else
 310                         Log(LogNotice, "IdoPgsqlConnection", "Empty program status table");
 311
 312                 /* if we did not write into the database earlier, another instance is active */
 313                 if (endpoint_name != my_endpoint->GetName()) {
 314                         double status_update_time;
 315
 316                         if (row)
 317                                 status_update_time = row->Get("status_update_time");
 318                         else
 319                                 status_update_time = 0;
 320
 321                         double status_update_age = Utility::GetTime() - status_update_time;
 322
 323                         Log(LogNotice, "IdoPgsqlConnection")
 324                                 << "Last update by '" << endpoint_name << "' was " << status_update_age << "s ago.";
 325
 326                         if (status_update_age < GetFailoverTimeout()) {
 327                                 m_Pgsql->finish(m_Connection);
 328                                 SetConnected(false);
 329                                 SetShouldConnect(false);
 330
 331                                 return;
 332                         }
 333
 334                         /* activate the IDO only, if we're authoritative in this zone */
 335                         if (IsPaused()) {
 336                                 Log(LogNotice, "IdoPgsqlConnection")
 337                                         << "Local endpoint '" << my_endpoint->GetName() << "' is not authoritative, bailing out.";
 338
 339                                 m_Pgsql->finish(m_Connection);
 340                                 SetConnected(false);
 341
 342                                 return;
 343                         }
 344                 }
 345
 346                 Log(LogNotice, "IdoPgsqlConnection", "Enabling IDO connection.");
 347         }
 348
 349         Log(LogInformation, "IdoPgsqlConnection")
 350                 << "pgSQL IDO instance id: " << static_cast<long>(m_InstanceID) << " (schema version: '" + version + "')";
 351
 352         Query("BEGIN");
 353
 354         /* update programstatus table */
 355         UpdateProgramStatus();
 356
 357         /* record connection */
 358         Query("INSERT INTO " + GetTablePrefix() + "conninfo " +
 359                 "(instance_id, connect_time, last_checkin_time, agent_name, agent_version, connect_type, data_start_time) VALUES ("
 360                 + Convert::ToString(static_cast<long>(m_InstanceID)) + ", NOW(), NOW(), E'icinga2 db_ido_pgsql', E'" + Escape(Application::GetAppVersion())
 361                 + "', E'" + (reconnect ? "RECONNECT" : "INITIAL") + "', NOW())");
 362
 363         /* clear config tables for the initial config dump */
 364         PrepareDatabase();
 365
 366         std::ostringstream q1buf;
 367         q1buf << "SELECT object_id, objecttype_id, name1, name2, is_active FROM " + GetTablePrefix() + "objects WHERE instance_id = " << static_cast<long>(m_InstanceID);
 368         result = Query(q1buf.str());
 369
 370         std::vector<DbObject::Ptr> activeDbObjs;
 371
 372         int index = 0;
 373         while ((row = FetchRow(result, index))) {
 374                 index++;
 375
 376                 DbType::Ptr dbtype = DbType::GetByID(row->Get("objecttype_id"));
 377
 378                 if (!dbtype)
 379                         continue;
 380
 381                 DbObject::Ptr dbobj = dbtype->GetOrCreateObjectByName(row->Get("name1"), row->Get("name2"));
 382                 SetObjectID(dbobj, DbReference(row->Get("object_id")));
 383                 bool active = row->Get("is_active");
 384                 SetObjectActive(dbobj, active);
 385
 386                 if (active)
 387                         activeDbObjs.push_back(dbobj);
 388         }
 389
 390         SetIDCacheValid(true);
 391
 392         EnableActiveChangedHandler();
 393
 394         for (const DbObject::Ptr& dbobj : activeDbObjs) {
 395                 if (dbobj->GetObject())
 396                         continue;
 397
 398                 Log(LogNotice, "IdoPgsqlConnection")
 399                         << "Deactivate deleted object name1: '" << dbobj->GetName1()
 400                         << "' name2: '" << dbobj->GetName2() + "'.";
 401                 DeactivateObject(dbobj);
 402         }
 403
 404         UpdateAllObjects();
 405
 406         m_QueryQueue.Enqueue(std::bind(&IdoPgsqlConnection::ClearTablesBySession, this), PriorityLow);
 407
 408         m_QueryQueue.Enqueue(std::bind(&IdoPgsqlConnection::FinishConnect, this, startTime), PriorityLow);
 409 }
 410
 411 void IdoPgsqlConnection::FinishConnect(double startTime)
 412 {
 413         AssertOnWorkQueue();
 414
 415         if (!GetConnected())
 416                 return;
 417
 418         Log(LogInformation, "IdoPgsqlConnection")
 419                 << "Finished reconnecting to PostgreSQL IDO database in " << std::setw(2) << Utility::GetTime() - startTime << " second(s).";
 420
 421         Query("COMMIT");
 422         Query("BEGIN");
 423 }
 424
 425 void IdoPgsqlConnection::ClearTablesBySession()
 426 {
 427         /* delete all comments and downtimes without current session token */
 428         ClearTableBySession("comments");
 429         ClearTableBySession("scheduleddowntime");
 430 }
 431
 432 void IdoPgsqlConnection::ClearTableBySession(const String& table)
 433 {
 434         Query("DELETE FROM " + GetTablePrefix() + table + " WHERE instance_id = " +
 435                 Convert::ToString(static_cast<long>(m_InstanceID)) + " AND session_token <> " +
 436                 Convert::ToString(GetSessionToken()));
 437 }
 438
 439 IdoPgsqlResult IdoPgsqlConnection::Query(const String& query)
 440 {
 441         AssertOnWorkQueue();
 442
 443         Log(LogDebug, "IdoPgsqlConnection")
 444                 << "Query: " << query;
 445
 446         IncreaseQueryCount();
 447
 448         PGresult *result = m_Pgsql->exec(m_Connection, query.CStr());
 449
 450         if (!result) {
 451                 String message = m_Pgsql->errorMessage(m_Connection);
 452                 Log(LogCritical, "IdoPgsqlConnection")
 453                         << "Error \"" << message << "\" when executing query \"" << query << "\"";
 454
 455                 BOOST_THROW_EXCEPTION(
 456                         database_error()
 457                         << errinfo_message(message)
 458                         << errinfo_database_query(query)
 459                 );
 460         }
 461
 462         char *rowCount = m_Pgsql->cmdTuples(result);
 463         m_AffectedRows = atoi(rowCount);
 464
 465         if (m_Pgsql->resultStatus(result) == PGRES_COMMAND_OK) {
 466                 m_Pgsql->clear(result);
 467                 return IdoPgsqlResult();
 468         }
 469
 470         if (m_Pgsql->resultStatus(result) != PGRES_TUPLES_OK) {
 471                 String message = m_Pgsql->resultErrorMessage(result);
 472                 m_Pgsql->clear(result);
 473
 474                 Log(LogCritical, "IdoPgsqlConnection")
 475                         << "Error \"" << message << "\" when executing query \"" << query << "\"";
 476
 477                 BOOST_THROW_EXCEPTION(
 478                         database_error()
 479                         << errinfo_message(message)
 480                         << errinfo_database_query(query)
 481                 );
 482         }
 483
 484         return IdoPgsqlResult(result, std::bind(&PgsqlInterface::clear, std::cref(m_Pgsql), _1));
 485 }
 486
 487 DbReference IdoPgsqlConnection::GetSequenceValue(const String& table, const String& column)
 488 {
 489         AssertOnWorkQueue();
 490
 491         IdoPgsqlResult result = Query("SELECT CURRVAL(pg_get_serial_sequence(E'" + Escape(table) + "', E'" + Escape(column) + "')) AS id");
 492
 493         Dictionary::Ptr row = FetchRow(result, 0);
 494
 495         ASSERT(row);
 496
 497         Log(LogDebug, "IdoPgsqlConnection")
 498                 << "Sequence Value: " << row->Get("id");
 499
 500         return {Convert::ToLong(row->Get("id"))};
 501 }
 502
 503 int IdoPgsqlConnection::GetAffectedRows()
 504 {
 505         AssertOnWorkQueue();
 506
 507         return m_AffectedRows;
 508 }
 509
 510 String IdoPgsqlConnection::Escape(const String& s)
 511 {
 512         AssertOnWorkQueue();
 513
 514         String utf8s = Utility::ValidateUTF8(s);
 515
 516         size_t length = utf8s.GetLength();
 517         auto *to = new char[utf8s.GetLength() * 2 + 1];
 518
 519         m_Pgsql->escapeStringConn(m_Connection, to, utf8s.CStr(), length, nullptr);
 520
 521         String result = String(to);
 522
 523         delete [] to;
 524
 525         return result;
 526 }
 527
 528 Dictionary::Ptr IdoPgsqlConnection::FetchRow(const IdoPgsqlResult& result, int row)
 529 {
 530         AssertOnWorkQueue();
 531
 532         if (row >= m_Pgsql->ntuples(result.get()))
 533                 return nullptr;
 534
 535         int columns = m_Pgsql->nfields(result.get());
 536
 537         Dictionary::Ptr dict = new Dictionary();
 538
 539         for (int column = 0; column < columns; column++) {
 540                 Value value;
 541
 542                 if (!m_Pgsql->getisnull(result.get(), row, column))
 543                         value = m_Pgsql->getvalue(result.get(), row, column);
 544
 545                 dict->Set(m_Pgsql->fname(result.get(), column), value);
 546         }
 547
 548         return dict;
 549 }
 550
 551 void IdoPgsqlConnection::ActivateObject(const DbObject::Ptr& dbobj)
 552 {
 553         m_QueryQueue.Enqueue(std::bind(&IdoPgsqlConnection::InternalActivateObject, this, dbobj), PriorityLow);
 554 }
 555
 556 void IdoPgsqlConnection::InternalActivateObject(const DbObject::Ptr& dbobj)
 557 {
 558         AssertOnWorkQueue();
 559
 560         if (!GetConnected())
 561                 return;
 562
 563         DbReference dbref = GetObjectID(dbobj);
 564         std::ostringstream qbuf;
 565
 566         if (!dbref.IsValid()) {
 567                 if (!dbobj->GetName2().IsEmpty()) {
 568                         qbuf << "INSERT INTO " + GetTablePrefix() + "objects (instance_id, objecttype_id, name1, name2, is_active) VALUES ("
 569                                 << static_cast<long>(m_InstanceID) << ", " << dbobj->GetType()->GetTypeID() << ", "
 570                                 << "E'" << Escape(dbobj->GetName1()) << "', E'" << Escape(dbobj->GetName2()) << "', 1)";
 571                 } else {
 572                         qbuf << "INSERT INTO " + GetTablePrefix() + "objects (instance_id, objecttype_id, name1, is_active) VALUES ("
 573                                 << static_cast<long>(m_InstanceID) << ", " << dbobj->GetType()->GetTypeID() << ", "
 574                                 << "E'" << Escape(dbobj->GetName1()) << "', 1)";
 575                 }
 576
 577                 Query(qbuf.str());
 578                 SetObjectID(dbobj, GetSequenceValue(GetTablePrefix() + "objects", "object_id"));
 579         } else {
 580                 qbuf << "UPDATE " + GetTablePrefix() + "objects SET is_active = 1 WHERE object_id = " << static_cast<long>(dbref);
 581                 Query(qbuf.str());
 582         }
 583 }
 584
 585 void IdoPgsqlConnection::DeactivateObject(const DbObject::Ptr& dbobj)
 586 {
 587         m_QueryQueue.Enqueue(std::bind(&IdoPgsqlConnection::InternalDeactivateObject, this, dbobj), PriorityLow);
 588 }
 589
 590 void IdoPgsqlConnection::InternalDeactivateObject(const DbObject::Ptr& dbobj)
 591 {
 592         AssertOnWorkQueue();
 593
 594         if (!GetConnected())
 595                 return;
 596
 597         DbReference dbref = GetObjectID(dbobj);
 598
 599         if (!dbref.IsValid())
 600                 return;
 601
 602         std::ostringstream qbuf;
 603         qbuf << "UPDATE " + GetTablePrefix() + "objects SET is_active = 0 WHERE object_id = " << static_cast<long>(dbref);
 604         Query(qbuf.str());
 605
 606         /* Note that we're _NOT_ clearing the db refs via SetReference/SetConfigUpdate/SetStatusUpdate
 607          * because the object is still in the database. */
 608 }
 609
 610 bool IdoPgsqlConnection::FieldToEscapedString(const String& key, const Value& value, Value *result)
 611 {
 612         if (key == "instance_id") {
 613                 *result = static_cast<long>(m_InstanceID);
 614                 return true;
 615         } else if (key == "session_token") {
 616                 *result = GetSessionToken();
 617                 return true;
 618         }
 619
 620         Value rawvalue = DbValue::ExtractValue(value);
 621
 622         if (rawvalue.IsObjectType<ConfigObject>()) {
 623                 DbObject::Ptr dbobjcol = DbObject::GetOrCreateByObject(rawvalue);
 624
 625                 if (!dbobjcol) {
 626                         *result = 0;
 627                         return true;
 628                 }
 629
 630                 if (!IsIDCacheValid())
 631                         return false;
 632
 633                 DbReference dbrefcol;
 634
 635                 if (DbValue::IsObjectInsertID(value)) {
 636                         dbrefcol = GetInsertID(dbobjcol);
 637
 638                         if (!dbrefcol.IsValid())
 639                                 return false;
 640                 } else {
 641                         dbrefcol = GetObjectID(dbobjcol);
 642
 643                         if (!dbrefcol.IsValid()) {
 644                                 InternalActivateObject(dbobjcol);
 645
 646                                 dbrefcol = GetObjectID(dbobjcol);
 647
 648                                 if (!dbrefcol.IsValid())
 649                                         return false;
 650                         }
 651                 }
 652
 653                 *result = static_cast<long>(dbrefcol);
 654         } else if (DbValue::IsTimestamp(value)) {
 655                 long ts = rawvalue;
 656                 std::ostringstream msgbuf;
 657                 msgbuf << "TO_TIMESTAMP(" << ts << ") AT TIME ZONE 'UTC'";
 658                 *result = Value(msgbuf.str());
 659         } else if (DbValue::IsTimestampNow(value)) {
 660                 *result = "NOW()";
 661         } else if (DbValue::IsObjectInsertID(value)) {
 662                 auto id = static_cast<long>(rawvalue);
 663
 664                 if (id <= 0)
 665                         return false;
 666
 667                 *result = id;
 668                 return true;
 669         } else {
 670                 Value fvalue;
 671
 672                 if (rawvalue.IsBoolean())
 673                         fvalue = Convert::ToLong(rawvalue);
 674                 else
 675                         fvalue = rawvalue;
 676
 677                 *result = "E'" + Escape(fvalue) + "'";
 678         }
 679
 680         return true;
 681 }
 682
 683 void IdoPgsqlConnection::ExecuteQuery(const DbQuery& query)
 684 {
 685         ASSERT(query.Category != DbCatInvalid);
 686
 687         m_QueryQueue.Enqueue(std::bind(&IdoPgsqlConnection::InternalExecuteQuery, this, query, -1), query.Priority, true);
 688 }
 689
 690 void IdoPgsqlConnection::ExecuteMultipleQueries(const std::vector<DbQuery>& queries)
 691 {
 692         if (queries.empty())
 693                 return;
 694
 695         m_QueryQueue.Enqueue(std::bind(&IdoPgsqlConnection::InternalExecuteMultipleQueries, this, queries), queries[0].Priority, true);
 696 }
 697
 698 bool IdoPgsqlConnection::CanExecuteQuery(const DbQuery& query)
 699 {
 700         if (query.Object && !IsIDCacheValid())
 701                 return false;
 702
 703         if (query.WhereCriteria) {
 704                 ObjectLock olock(query.WhereCriteria);
 705                 Value value;
 706
 707                 for (const Dictionary::Pair& kv : query.WhereCriteria) {
 708                         if (!FieldToEscapedString(kv.first, kv.second, &value))
 709                                 return false;
 710                 }
 711         }
 712
 713         if (query.Fields) {
 714                 ObjectLock olock(query.Fields);
 715
 716                 for (const Dictionary::Pair& kv : query.Fields) {
 717                         Value value;
 718
 719                         if (kv.second.IsEmpty() && !kv.second.IsString())
 720                                 continue;
 721
 722                         if (!FieldToEscapedString(kv.first, kv.second, &value))
 723                                 return false;
 724                 }
 725         }
 726
 727         return true;
 728 }
 729
 730 void IdoPgsqlConnection::InternalExecuteMultipleQueries(const std::vector<DbQuery>& queries)
 731 {
 732         AssertOnWorkQueue();
 733
 734         if (!GetConnected())
 735                 return;
 736
 737         for (const DbQuery& query : queries) {
 738                 ASSERT(query.Type == DbQueryNewTransaction || query.Category != DbCatInvalid);
 739
 740                 if (!CanExecuteQuery(query)) {
 741                         m_QueryQueue.Enqueue(std::bind(&IdoPgsqlConnection::InternalExecuteMultipleQueries, this, queries), query.Priority);
 742                         return;
 743                 }
 744         }
 745
 746         for (const DbQuery& query : queries) {
 747                 InternalExecuteQuery(query);
 748         }
 749 }
 750
 751 void IdoPgsqlConnection::InternalExecuteQuery(const DbQuery& query, int typeOverride)
 752 {
 753         AssertOnWorkQueue();
 754
 755         if (!GetConnected())
 756                 return;
 757
 758         if (query.Type == DbQueryNewTransaction) {
 759                 InternalNewTransaction();
 760                 return;
 761         }
 762
 763         /* check whether we're allowed to execute the query first */
 764         if (GetCategoryFilter() != DbCatEverything && (query.Category & GetCategoryFilter()) == 0)
 765                 return;
 766
 767         if (query.Object && query.Object->GetObject()->GetExtension("agent_check").ToBool())
 768                 return;
 769
 770         /* check if there are missing object/insert ids and re-enqueue the query */
 771         if (!CanExecuteQuery(query)) {
 772                 m_QueryQueue.Enqueue(std::bind(&IdoPgsqlConnection::InternalExecuteQuery, this, query, typeOverride), query.Priority);
 773                 return;
 774         }
 775
 776         std::ostringstream qbuf, where;
 777         int type;
 778
 779         if (query.WhereCriteria) {
 780                 where << " WHERE ";
 781
 782                 ObjectLock olock(query.WhereCriteria);
 783                 Value value;
 784                 bool first = true;
 785
 786                 for (const Dictionary::Pair& kv : query.WhereCriteria) {
 787                         if (!FieldToEscapedString(kv.first, kv.second, &value)) {
 788                                 m_QueryQueue.Enqueue(std::bind(&IdoPgsqlConnection::InternalExecuteQuery, this, query, -1), query.Priority);
 789                                 return;
 790                         }
 791
 792                         if (!first)
 793                                 where << " AND ";
 794
 795                         where << kv.first << " = " << value;
 796
 797                         if (first)
 798                                 first = false;
 799                 }
 800         }
 801
 802         type = (typeOverride != -1) ? typeOverride : query.Type;
 803
 804         bool upsert = false;
 805
 806         if ((type & DbQueryInsert) && (type & DbQueryUpdate)) {
 807                 bool hasid = false;
 808
 809                 if (query.Object) {
 810                         if (query.ConfigUpdate)
 811                                 hasid = GetConfigUpdate(query.Object);
 812                         else if (query.StatusUpdate)
 813                                 hasid = GetStatusUpdate(query.Object);
 814                 }
 815
 816                 if (!hasid)
 817                         upsert = true;
 818
 819                 type = DbQueryUpdate;
 820         }
 821
 822         if ((type & DbQueryInsert) && (type & DbQueryDelete)) {
 823                 std::ostringstream qdel;
 824                 qdel << "DELETE FROM " << GetTablePrefix() << query.Table << where.str();
 825                 Query(qdel.str());
 826
 827                 type = DbQueryInsert;
 828         }
 829
 830         switch (type) {
 831                 case DbQueryInsert:
 832                         qbuf << "INSERT INTO " << GetTablePrefix() << query.Table;
 833                         break;
 834                 case DbQueryUpdate:
 835                         qbuf << "UPDATE " << GetTablePrefix() << query.Table << " SET";
 836                         break;
 837                 case DbQueryDelete:
 838                         qbuf << "DELETE FROM " << GetTablePrefix() << query.Table;
 839                         break;
 840                 default:
 841                         VERIFY(!"Invalid query type.");
 842         }
 843
 844         if (type == DbQueryInsert || type == DbQueryUpdate) {
 845                 std::ostringstream colbuf, valbuf;
 846
 847                 if (type == DbQueryUpdate && query.Fields->GetLength() == 0)
 848                         return;
 849
 850                 ObjectLock olock(query.Fields);
 851
 852                 Value value;
 853                 bool first = true;
 854                 for (const Dictionary::Pair& kv : query.Fields) {
 855                         if (kv.second.IsEmpty() && !kv.second.IsString())
 856                                 continue;
 857
 858                         if (!FieldToEscapedString(kv.first, kv.second, &value)) {
 859                                 m_QueryQueue.Enqueue(std::bind(&IdoPgsqlConnection::InternalExecuteQuery, this, query, -1), query.Priority);
 860                                 return;
 861                         }
 862
 863                         if (type == DbQueryInsert) {
 864                                 if (!first) {
 865                                         colbuf << ", ";
 866                                         valbuf << ", ";
 867                                 }
 868
 869                                 colbuf << kv.first;
 870                                 valbuf << value;
 871                         } else {
 872                                 if (!first)
 873                                         qbuf << ", ";
 874
 875                                 qbuf << " " << kv.first << " = " << value;
 876                         }
 877
 878                         if (first)
 879                                 first = false;
 880                 }
 881
 882                 if (type == DbQueryInsert)
 883                         qbuf << " (" << colbuf.str() << ") VALUES (" << valbuf.str() << ")";
 884         }
 885
 886         if (type != DbQueryInsert)
 887                 qbuf << where.str();
 888
 889         Query(qbuf.str());
 890
 891         if (upsert && GetAffectedRows() == 0) {
 892                 InternalExecuteQuery(query, DbQueryDelete | DbQueryInsert);
 893
 894                 return;
 895         }
 896
 897         if (type == DbQueryInsert && query.Object) {
 898                 if (query.ConfigUpdate) {
 899                         String idField = query.IdColumn;
 900
 901                         if (idField.IsEmpty())
 902                                 idField = query.Table.SubStr(0, query.Table.GetLength() - 1) + "_id";
 903
 904                         SetInsertID(query.Object, GetSequenceValue(GetTablePrefix() + query.Table, idField));
 905
 906                         SetConfigUpdate(query.Object, true);
 907                 } else if (query.StatusUpdate)
 908                         SetStatusUpdate(query.Object, true);
 909         }
 910
 911         if (type == DbQueryInsert && query.Table == "notifications" && query.NotificationInsertID) {
 912                 DbReference seqval = GetSequenceValue(GetTablePrefix() + query.Table, "notification_id");
 913                 query.NotificationInsertID->SetValue(static_cast<long>(seqval));
 914         }
 915 }
 916
 917 void IdoPgsqlConnection::CleanUpExecuteQuery(const String& table, const String& time_column, double max_age)
 918 {
 919         m_QueryQueue.Enqueue(std::bind(&IdoPgsqlConnection::InternalCleanUpExecuteQuery, this, table, time_column, max_age), PriorityLow, true);
 920 }
 921
 922 void IdoPgsqlConnection::InternalCleanUpExecuteQuery(const String& table, const String& time_column, double max_age)
 923 {
 924         AssertOnWorkQueue();
 925
 926         if (!GetConnected())
 927                 return;
 928
 929         Query("DELETE FROM " + GetTablePrefix() + table + " WHERE instance_id = " +
 930                 Convert::ToString(static_cast<long>(m_InstanceID)) + " AND " + time_column +
 931                 " < TO_TIMESTAMP(" + Convert::ToString(static_cast<long>(max_age)) + ")");
 932 }
 933
 934 void IdoPgsqlConnection::FillIDCache(const DbType::Ptr& type)
 935 {
 936         String query = "SELECT " + type->GetIDColumn() + " AS object_id, " + type->GetTable() + "_id, config_hash FROM " + GetTablePrefix() + type->GetTable() + "s";
 937         IdoPgsqlResult result = Query(query);
 938
 939         Dictionary::Ptr row;
 940
 941         int index = 0;
 942         while ((row = FetchRow(result, index))) {
 943                 index++;
 944                 DbReference dbref(row->Get("object_id"));
 945                 SetInsertID(type, dbref, DbReference(row->Get(type->GetTable() + "_id")));
 946                 SetConfigHash(type, dbref, row->Get("config_hash"));
 947         }
 948 }
 949
 950 int IdoPgsqlConnection::GetPendingQueryCount() const
 951 {
 952         return m_QueryQueue.GetLength();
 953 }