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