diff options
Diffstat (limited to 'lib/db_ido_mysql/idomysqlconnection.cpp')
| -rw-r--r-- | lib/db_ido_mysql/idomysqlconnection.cpp | 1269 |
1 files changed, 1269 insertions, 0 deletions
diff --git a/lib/db_ido_mysql/idomysqlconnection.cpp b/lib/db_ido_mysql/idomysqlconnection.cpp new file mode 100644 index 0000000..aacb7d7 --- /dev/null +++ b/lib/db_ido_mysql/idomysqlconnection.cpp @@ -0,0 +1,1269 @@ +/* Icinga 2 | (c) 2012 Icinga GmbH | GPLv2+ */ + +#include "db_ido_mysql/idomysqlconnection.hpp" +#include "db_ido_mysql/idomysqlconnection-ti.cpp" +#include "db_ido/dbtype.hpp" +#include "db_ido/dbvalue.hpp" +#include "base/logger.hpp" +#include "base/objectlock.hpp" +#include "base/convert.hpp" +#include "base/utility.hpp" +#include "base/perfdatavalue.hpp" +#include "base/application.hpp" +#include "base/configtype.hpp" +#include "base/exception.hpp" +#include "base/statsfunction.hpp" +#include "base/defer.hpp" +#include <utility> + +using namespace icinga; + +REGISTER_TYPE(IdoMysqlConnection); +REGISTER_STATSFUNCTION(IdoMysqlConnection, &IdoMysqlConnection::StatsFunc); + +const char * IdoMysqlConnection::GetLatestSchemaVersion() const noexcept +{ + return "1.15.1"; +} + +const char * IdoMysqlConnection::GetCompatSchemaVersion() const noexcept +{ + return "1.14.3"; +} + +void IdoMysqlConnection::OnConfigLoaded() +{ + ObjectImpl<IdoMysqlConnection>::OnConfigLoaded(); + + m_QueryQueue.SetName("IdoMysqlConnection, " + GetName()); + + Library shimLibrary{"mysql_shim"}; + + auto create_mysql_shim = shimLibrary.GetSymbolAddress<create_mysql_shim_ptr>("create_mysql_shim"); + + m_Mysql.reset(create_mysql_shim()); + + std::swap(m_Library, shimLibrary); +} + +void IdoMysqlConnection::StatsFunc(const Dictionary::Ptr& status, const Array::Ptr& perfdata) +{ + DictionaryData nodes; + + for (const IdoMysqlConnection::Ptr& idomysqlconnection : ConfigType::GetObjectsByType<IdoMysqlConnection>()) { + size_t queryQueueItems = idomysqlconnection->m_QueryQueue.GetLength(); + double queryQueueItemRate = idomysqlconnection->m_QueryQueue.GetTaskCount(60) / 60.0; + + nodes.emplace_back(idomysqlconnection->GetName(), new Dictionary({ + { "version", idomysqlconnection->GetSchemaVersion() }, + { "instance_name", idomysqlconnection->GetInstanceName() }, + { "connected", idomysqlconnection->GetConnected() }, + { "query_queue_items", queryQueueItems }, + { "query_queue_item_rate", queryQueueItemRate } + })); + + perfdata->Add(new PerfdataValue("idomysqlconnection_" + idomysqlconnection->GetName() + "_queries_rate", idomysqlconnection->GetQueryCount(60) / 60.0)); + perfdata->Add(new PerfdataValue("idomysqlconnection_" + idomysqlconnection->GetName() + "_queries_1min", idomysqlconnection->GetQueryCount(60))); + perfdata->Add(new PerfdataValue("idomysqlconnection_" + idomysqlconnection->GetName() + "_queries_5mins", idomysqlconnection->GetQueryCount(5 * 60))); + perfdata->Add(new PerfdataValue("idomysqlconnection_" + idomysqlconnection->GetName() + "_queries_15mins", idomysqlconnection->GetQueryCount(15 * 60))); + perfdata->Add(new PerfdataValue("idomysqlconnection_" + idomysqlconnection->GetName() + "_query_queue_items", queryQueueItems)); + perfdata->Add(new PerfdataValue("idomysqlconnection_" + idomysqlconnection->GetName() + "_query_queue_item_rate", queryQueueItemRate)); + } + + status->Set("idomysqlconnection", new Dictionary(std::move(nodes))); +} + +void IdoMysqlConnection::Resume() +{ + Log(LogInformation, "IdoMysqlConnection") + << "'" << GetName() << "' resumed."; + + SetConnected(false); + + m_QueryQueue.SetExceptionCallback([this](boost::exception_ptr exp) { ExceptionHandler(std::move(exp)); }); + + /* Immediately try to connect on Resume() without timer. */ + m_QueryQueue.Enqueue([this]() { Reconnect(); }, PriorityImmediate); + + m_TxTimer = Timer::Create(); + m_TxTimer->SetInterval(1); + m_TxTimer->OnTimerExpired.connect([this](const Timer * const&) { NewTransaction(); }); + m_TxTimer->Start(); + + m_ReconnectTimer = Timer::Create(); + m_ReconnectTimer->SetInterval(10); + m_ReconnectTimer->OnTimerExpired.connect([this](const Timer * const&){ ReconnectTimerHandler(); }); + m_ReconnectTimer->Start(); + + /* Start with queries after connect. */ + DbConnection::Resume(); + + ASSERT(m_Mysql->thread_safe()); +} + +void IdoMysqlConnection::Pause() +{ + Log(LogDebug, "IdoMysqlConnection") + << "Attempting to pause '" << GetName() << "'."; + + DbConnection::Pause(); + + m_ReconnectTimer->Stop(true); + m_TxTimer->Stop(true); + +#ifdef I2_DEBUG /* I2_DEBUG */ + Log(LogDebug, "IdoMysqlConnection") + << "Rescheduling disconnect task."; +#endif /* I2_DEBUG */ + + Log(LogInformation, "IdoMysqlConnection") + << "'" << GetName() << "' paused."; + +} + +void IdoMysqlConnection::ExceptionHandler(boost::exception_ptr exp) +{ + Log(LogCritical, "IdoMysqlConnection", "Exception during database operation: Verify that your database is operational!"); + + Log(LogDebug, "IdoMysqlConnection") + << "Exception during database operation: " << DiagnosticInformation(std::move(exp)); + + if (GetConnected()) { + m_Mysql->close(&m_Connection); + + SetConnected(false); + } +} + +void IdoMysqlConnection::AssertOnWorkQueue() +{ + ASSERT(m_QueryQueue.IsWorkerThread()); +} + +void IdoMysqlConnection::Disconnect() +{ + AssertOnWorkQueue(); + + if (!GetConnected()) + return; + + Query("COMMIT"); + m_Mysql->close(&m_Connection); + + SetConnected(false); + + Log(LogInformation, "IdoMysqlConnection") + << "Disconnected from '" << GetName() << "' database '" << GetDatabase() << "'."; +} + +void IdoMysqlConnection::NewTransaction() +{ + if (IsPaused() && GetPauseCalled()) + return; + +#ifdef I2_DEBUG /* I2_DEBUG */ + Log(LogDebug, "IdoMysqlConnection") + << "Scheduling new transaction and finishing async queries."; +#endif /* I2_DEBUG */ + + m_QueryQueue.Enqueue([this]() { InternalNewTransaction(); }, PriorityHigh); +} + +void IdoMysqlConnection::InternalNewTransaction() +{ + AssertOnWorkQueue(); + + if (!GetConnected()) + return; + + IncreasePendingQueries(2); + + AsyncQuery("COMMIT"); + AsyncQuery("BEGIN"); + + FinishAsyncQueries(); +} + +void IdoMysqlConnection::ReconnectTimerHandler() +{ +#ifdef I2_DEBUG /* I2_DEBUG */ + Log(LogDebug, "IdoMysqlConnection") + << "Scheduling reconnect task."; +#endif /* I2_DEBUG */ + + /* Only allow Reconnect events with high priority. */ + m_QueryQueue.Enqueue([this]() { Reconnect(); }, PriorityImmediate); +} + +void IdoMysqlConnection::Reconnect() +{ + AssertOnWorkQueue(); + + if (!IsActive()) + return; + + CONTEXT("Reconnecting to MySQL IDO database '" << GetName() << "'"); + + double startTime = Utility::GetTime(); + + SetShouldConnect(true); + + bool reconnect = false; + + /* Ensure to close old connections first. */ + if (GetConnected()) { + /* Check if we're really still connected */ + if (m_Mysql->ping(&m_Connection) == 0) + return; + + m_Mysql->close(&m_Connection); + SetConnected(false); + reconnect = true; + } + + Log(LogDebug, "IdoMysqlConnection") + << "Reconnect: Clearing ID cache."; + + ClearIDCache(); + + String ihost, isocket_path, iuser, ipasswd, idb; + String isslKey, isslCert, isslCa, isslCaPath, isslCipher; + const char *host, *socket_path, *user , *passwd, *db; + const char *sslKey, *sslCert, *sslCa, *sslCaPath, *sslCipher; + bool enableSsl; + long port; + + ihost = GetHost(); + isocket_path = GetSocketPath(); + iuser = GetUser(); + ipasswd = GetPassword(); + idb = GetDatabase(); + + enableSsl = GetEnableSsl(); + isslKey = GetSslKey(); + isslCert = GetSslCert(); + isslCa = GetSslCa(); + isslCaPath = GetSslCapath(); + isslCipher = GetSslCipher(); + + host = (!ihost.IsEmpty()) ? ihost.CStr() : nullptr; + port = GetPort(); + socket_path = (!isocket_path.IsEmpty()) ? isocket_path.CStr() : nullptr; + user = (!iuser.IsEmpty()) ? iuser.CStr() : nullptr; + passwd = (!ipasswd.IsEmpty()) ? ipasswd.CStr() : nullptr; + db = (!idb.IsEmpty()) ? idb.CStr() : nullptr; + + sslKey = (!isslKey.IsEmpty()) ? isslKey.CStr() : nullptr; + sslCert = (!isslCert.IsEmpty()) ? isslCert.CStr() : nullptr; + sslCa = (!isslCa.IsEmpty()) ? isslCa.CStr() : nullptr; + sslCaPath = (!isslCaPath.IsEmpty()) ? isslCaPath.CStr() : nullptr; + sslCipher = (!isslCipher.IsEmpty()) ? isslCipher.CStr() : nullptr; + + /* connection */ + if (!m_Mysql->init(&m_Connection)) { + Log(LogCritical, "IdoMysqlConnection") + << "mysql_init() failed: out of memory"; + + BOOST_THROW_EXCEPTION(std::bad_alloc()); + } + + /* Read "latin1" (here, in the schema and in Icinga Web) as "bytes". + Icinga 2 and Icinga Web use byte-strings everywhere and every byte-string is a valid latin1 string. + This way the (actually mostly UTF-8) bytes are transferred end-to-end as-is. */ + m_Mysql->options(&m_Connection, MYSQL_SET_CHARSET_NAME, "latin1"); + + if (enableSsl) + m_Mysql->ssl_set(&m_Connection, sslKey, sslCert, sslCa, sslCaPath, sslCipher); + + if (!m_Mysql->real_connect(&m_Connection, host, user, passwd, db, port, socket_path, CLIENT_FOUND_ROWS | CLIENT_MULTI_STATEMENTS)) { + Log(LogCritical, "IdoMysqlConnection") + << "Connection to database '" << db << "' with user '" << user << "' on '" << host << ":" << port + << "' " << (enableSsl ? "(SSL enabled) " : "") << "failed: \"" << m_Mysql->error(&m_Connection) << "\""; + + BOOST_THROW_EXCEPTION(std::runtime_error(m_Mysql->error(&m_Connection))); + } + + Log(LogNotice, "IdoMysqlConnection") + << "Reconnect: '" << GetName() << "' is now connected to database '" << GetDatabase() << "'."; + + SetConnected(true); + + IdoMysqlResult result = Query("SELECT @@global.max_allowed_packet AS max_allowed_packet"); + + Dictionary::Ptr row = FetchRow(result); + + if (row) + m_MaxPacketSize = row->Get("max_allowed_packet"); + else + m_MaxPacketSize = 64 * 1024; + + DiscardRows(result); + + String dbVersionName = "idoutils"; + result = Query("SELECT version FROM " + GetTablePrefix() + "dbversion WHERE name='" + Escape(dbVersionName) + "'"); + + row = FetchRow(result); + + if (!row) { + m_Mysql->close(&m_Connection); + SetConnected(false); + + Log(LogCritical, "IdoMysqlConnection", "Schema does not provide any valid version! Verify your schema installation."); + + BOOST_THROW_EXCEPTION(std::runtime_error("Invalid schema.")); + } + + DiscardRows(result); + + String version = row->Get("version"); + + SetSchemaVersion(version); + + if (Utility::CompareVersion(GetCompatSchemaVersion(), version) < 0) { + m_Mysql->close(&m_Connection); + SetConnected(false); + + Log(LogCritical, "IdoMysqlConnection") + << "Schema version '" << version << "' does not match the required version '" + << GetCompatSchemaVersion() << "' (or newer)! Please check the upgrade documentation at " + << "https://icinga.com/docs/icinga2/latest/doc/16-upgrading-icinga-2/#upgrading-mysql-db"; + + BOOST_THROW_EXCEPTION(std::runtime_error("Schema version mismatch.")); + } + + String instanceName = GetInstanceName(); + + result = Query("SELECT instance_id FROM " + GetTablePrefix() + "instances WHERE instance_name = '" + Escape(instanceName) + "'"); + row = FetchRow(result); + + if (!row) { + Query("INSERT INTO " + GetTablePrefix() + "instances (instance_name, instance_description) VALUES ('" + Escape(instanceName) + "', '" + Escape(GetInstanceDescription()) + "')"); + m_InstanceID = GetLastInsertID(); + } else { + m_InstanceID = DbReference(row->Get("instance_id")); + } + + DiscardRows(result); + + Endpoint::Ptr my_endpoint = Endpoint::GetLocalEndpoint(); + + /* we have an endpoint in a cluster setup, so decide if we can proceed here */ + if (my_endpoint && GetHAMode() == HARunOnce) { + /* get the current endpoint writing to programstatus table */ + result = Query("SELECT UNIX_TIMESTAMP(status_update_time) AS status_update_time, endpoint_name FROM " + + GetTablePrefix() + "programstatus WHERE instance_id = " + Convert::ToString(m_InstanceID)); + row = FetchRow(result); + DiscardRows(result); + + String endpoint_name; + + if (row) + endpoint_name = row->Get("endpoint_name"); + else + Log(LogNotice, "IdoMysqlConnection", "Empty program status table"); + + /* if we did not write into the database earlier, another instance is active */ + if (endpoint_name != my_endpoint->GetName()) { + double status_update_time; + + if (row) + status_update_time = row->Get("status_update_time"); + else + status_update_time = 0; + + double now = Utility::GetTime(); + + double status_update_age = now - status_update_time; + double failoverTimeout = GetFailoverTimeout(); + + if (status_update_age < failoverTimeout) { + Log(LogInformation, "IdoMysqlConnection") + << "Last update by endpoint '" << endpoint_name << "' was " + << status_update_age << "s ago (< failover timeout of " << failoverTimeout << "s). Retrying."; + + m_Mysql->close(&m_Connection); + SetConnected(false); + SetShouldConnect(false); + + return; + } + + /* activate the IDO only, if we're authoritative in this zone */ + if (IsPaused()) { + Log(LogNotice, "IdoMysqlConnection") + << "Local endpoint '" << my_endpoint->GetName() << "' is not authoritative, bailing out."; + + m_Mysql->close(&m_Connection); + SetConnected(false); + + return; + } + + SetLastFailover(now); + + Log(LogInformation, "IdoMysqlConnection") + << "Last update by endpoint '" << endpoint_name << "' was " + << status_update_age << "s ago. Taking over '" << GetName() << "' in HA zone '" << Zone::GetLocalZone()->GetName() << "'."; + } + + Log(LogNotice, "IdoMysqlConnection", "Enabling IDO connection in HA zone."); + } + + Log(LogInformation, "IdoMysqlConnection") + << "MySQL IDO instance id: " << static_cast<long>(m_InstanceID) << " (schema version: '" + version + "')"; + + /* set session time zone to utc */ + Query("SET SESSION TIME_ZONE='+00:00'"); + + Query("SET SESSION SQL_MODE='NO_AUTO_VALUE_ON_ZERO'"); + + Query("BEGIN"); + + /* update programstatus table */ + UpdateProgramStatus(); + + /* record connection */ + Query("INSERT INTO " + GetTablePrefix() + "conninfo " + + "(instance_id, connect_time, last_checkin_time, agent_name, agent_version, connect_type, data_start_time) VALUES (" + + Convert::ToString(static_cast<long>(m_InstanceID)) + ", NOW(), NOW(), 'icinga2 db_ido_mysql', '" + Escape(Application::GetAppVersion()) + + "', '" + (reconnect ? "RECONNECT" : "INITIAL") + "', NOW())"); + + /* clear config tables for the initial config dump */ + PrepareDatabase(); + + std::ostringstream q1buf; + q1buf << "SELECT object_id, objecttype_id, name1, name2, is_active FROM " + GetTablePrefix() + "objects WHERE instance_id = " << static_cast<long>(m_InstanceID); + result = Query(q1buf.str()); + + std::vector<DbObject::Ptr> activeDbObjs; + + while ((row = FetchRow(result))) { + DbType::Ptr dbtype = DbType::GetByID(row->Get("objecttype_id")); + + if (!dbtype) + continue; + + DbObject::Ptr dbobj = dbtype->GetOrCreateObjectByName(row->Get("name1"), row->Get("name2")); + SetObjectID(dbobj, DbReference(row->Get("object_id"))); + bool active = row->Get("is_active"); + SetObjectActive(dbobj, active); + + if (active) + activeDbObjs.emplace_back(std::move(dbobj)); + } + + SetIDCacheValid(true); + + EnableActiveChangedHandler(); + + for (const DbObject::Ptr& dbobj : activeDbObjs) { + if (dbobj->GetObject()) + continue; + + Log(LogNotice, "IdoMysqlConnection") + << "Deactivate deleted object name1: '" << dbobj->GetName1() + << "' name2: '" << dbobj->GetName2() + "'."; + DeactivateObject(dbobj); + } + + UpdateAllObjects(); + +#ifdef I2_DEBUG /* I2_DEBUG */ + Log(LogDebug, "IdoMysqlConnection") + << "Scheduling session table clear and finish connect task."; +#endif /* I2_DEBUG */ + + m_QueryQueue.Enqueue([this]() { ClearTablesBySession(); }, PriorityNormal); + + m_QueryQueue.Enqueue([this, startTime]() { FinishConnect(startTime); }, PriorityNormal); +} + +void IdoMysqlConnection::FinishConnect(double startTime) +{ + AssertOnWorkQueue(); + + if (!GetConnected() || IsPaused()) + return; + + FinishAsyncQueries(); + + Log(LogInformation, "IdoMysqlConnection") + << "Finished reconnecting to '" << GetName() << "' database '" << GetDatabase() << "' in " + << std::setw(2) << Utility::GetTime() - startTime << " second(s)."; + + Query("COMMIT"); + Query("BEGIN"); +} + +void IdoMysqlConnection::ClearTablesBySession() +{ + /* delete all comments and downtimes without current session token */ + ClearTableBySession("comments"); + ClearTableBySession("scheduleddowntime"); +} + +void IdoMysqlConnection::ClearTableBySession(const String& table) +{ + Query("DELETE FROM " + GetTablePrefix() + table + " WHERE instance_id = " + + Convert::ToString(static_cast<long>(m_InstanceID)) + " AND session_token <> " + + Convert::ToString(GetSessionToken())); +} + +void IdoMysqlConnection::AsyncQuery(const String& query, const std::function<void (const IdoMysqlResult&)>& callback) +{ + AssertOnWorkQueue(); + + IdoAsyncQuery aq; + aq.Query = query; + /* XXX: Important: The callback must not immediately execute a query, but enqueue it! + * See https://github.com/Icinga/icinga2/issues/4603 for details. + */ + aq.Callback = callback; + m_AsyncQueries.emplace_back(std::move(aq)); +} + +void IdoMysqlConnection::FinishAsyncQueries() +{ + std::vector<IdoAsyncQuery> queries; + m_AsyncQueries.swap(queries); + + std::vector<IdoAsyncQuery>::size_type offset = 0; + + // This will be executed if there is a problem with executing the queries, + // at which point this function throws an exception and the queries should + // not be listed as still pending in the queue. + Defer decreaseQueries ([this, &offset, &queries]() { + auto lostQueries = queries.size() - offset; + + if (lostQueries > 0) { + DecreasePendingQueries(lostQueries); + } + }); + + while (offset < queries.size()) { + std::ostringstream querybuf; + + std::vector<IdoAsyncQuery>::size_type count = 0; + size_t num_bytes = 0; + + Defer decreaseQueries ([this, &offset, &count]() { + offset += count; + DecreasePendingQueries(count); + m_UncommittedAsyncQueries += count; + }); + + for (std::vector<IdoAsyncQuery>::size_type i = offset; i < queries.size(); i++) { + const IdoAsyncQuery& aq = queries[i]; + + size_t size_query = aq.Query.GetLength() + 1; + + if (count > 0) { + if (num_bytes + size_query > m_MaxPacketSize - 512) + break; + + querybuf << ";"; + } + + IncreaseQueryCount(); + count++; + + Log(LogDebug, "IdoMysqlConnection") + << "Query: " << aq.Query; + + querybuf << aq.Query; + num_bytes += size_query; + } + + String query = querybuf.str(); + + if (m_Mysql->query(&m_Connection, query.CStr()) != 0) { + std::ostringstream msgbuf; + String message = m_Mysql->error(&m_Connection); + msgbuf << "Error \"" << message << "\" when executing query \"" << query << "\""; + Log(LogCritical, "IdoMysqlConnection", msgbuf.str()); + + BOOST_THROW_EXCEPTION( + database_error() + << errinfo_message(m_Mysql->error(&m_Connection)) + << errinfo_database_query(query) + ); + } + + for (std::vector<IdoAsyncQuery>::size_type i = offset; i < offset + count; i++) { + const IdoAsyncQuery& aq = queries[i]; + + MYSQL_RES *result = m_Mysql->store_result(&m_Connection); + + m_AffectedRows = m_Mysql->affected_rows(&m_Connection); + + IdoMysqlResult iresult; + + if (!result) { + if (m_Mysql->field_count(&m_Connection) > 0) { + std::ostringstream msgbuf; + String message = m_Mysql->error(&m_Connection); + msgbuf << "Error \"" << message << "\" when executing query \"" << aq.Query << "\""; + Log(LogCritical, "IdoMysqlConnection", msgbuf.str()); + + BOOST_THROW_EXCEPTION( + database_error() + << errinfo_message(m_Mysql->error(&m_Connection)) + << errinfo_database_query(query) + ); + } + } else + iresult = IdoMysqlResult(result, [this](MYSQL_RES* result) { m_Mysql->free_result(result); }); + + if (aq.Callback) + aq.Callback(iresult); + + if (m_Mysql->next_result(&m_Connection) > 0) { + std::ostringstream msgbuf; + String message = m_Mysql->error(&m_Connection); + msgbuf << "Error \"" << message << "\" when executing query \"" << query << "\""; + Log(LogCritical, "IdoMysqlConnection", msgbuf.str()); + + BOOST_THROW_EXCEPTION( + database_error() + << errinfo_message(m_Mysql->error(&m_Connection)) + << errinfo_database_query(query) + ); + } + } + } + + if (m_UncommittedAsyncQueries > 25000) { + m_UncommittedAsyncQueries = 0; + + Query("COMMIT"); + Query("BEGIN"); + } +} + +IdoMysqlResult IdoMysqlConnection::Query(const String& query) +{ + AssertOnWorkQueue(); + + IncreasePendingQueries(1); + Defer decreaseQueries ([this]() { DecreasePendingQueries(1); }); + + /* finish all async queries to maintain the right order for queries */ + FinishAsyncQueries(); + + Log(LogDebug, "IdoMysqlConnection") + << "Query: " << query; + + IncreaseQueryCount(); + + if (m_Mysql->query(&m_Connection, query.CStr()) != 0) { + std::ostringstream msgbuf; + String message = m_Mysql->error(&m_Connection); + msgbuf << "Error \"" << message << "\" when executing query \"" << query << "\""; + Log(LogCritical, "IdoMysqlConnection", msgbuf.str()); + + BOOST_THROW_EXCEPTION( + database_error() + << errinfo_message(m_Mysql->error(&m_Connection)) + << errinfo_database_query(query) + ); + } + + MYSQL_RES *result = m_Mysql->store_result(&m_Connection); + + m_AffectedRows = m_Mysql->affected_rows(&m_Connection); + + if (!result) { + if (m_Mysql->field_count(&m_Connection) > 0) { + std::ostringstream msgbuf; + String message = m_Mysql->error(&m_Connection); + msgbuf << "Error \"" << message << "\" when executing query \"" << query << "\""; + Log(LogCritical, "IdoMysqlConnection", msgbuf.str()); + + BOOST_THROW_EXCEPTION( + database_error() + << errinfo_message(m_Mysql->error(&m_Connection)) + << errinfo_database_query(query) + ); + } + + return IdoMysqlResult(); + } + + return IdoMysqlResult(result, [this](MYSQL_RES* result) { m_Mysql->free_result(result); }); +} + +DbReference IdoMysqlConnection::GetLastInsertID() +{ + AssertOnWorkQueue(); + + return {static_cast<long>(m_Mysql->insert_id(&m_Connection))}; +} + +int IdoMysqlConnection::GetAffectedRows() +{ + AssertOnWorkQueue(); + + return m_AffectedRows; +} + +String IdoMysqlConnection::Escape(const String& s) +{ + AssertOnWorkQueue(); + + String utf8s = Utility::ValidateUTF8(s); + + size_t length = utf8s.GetLength(); + auto *to = new char[utf8s.GetLength() * 2 + 1]; + + m_Mysql->real_escape_string(&m_Connection, to, utf8s.CStr(), length); + + String result = String(to); + + delete [] to; + + return result; +} + +Dictionary::Ptr IdoMysqlConnection::FetchRow(const IdoMysqlResult& result) +{ + AssertOnWorkQueue(); + + MYSQL_ROW row; + MYSQL_FIELD *field; + unsigned long *lengths, i; + + row = m_Mysql->fetch_row(result.get()); + + if (!row) + return nullptr; + + lengths = m_Mysql->fetch_lengths(result.get()); + + if (!lengths) + return nullptr; + + Dictionary::Ptr dict = new Dictionary(); + + m_Mysql->field_seek(result.get(), 0); + for (field = m_Mysql->fetch_field(result.get()), i = 0; field; field = m_Mysql->fetch_field(result.get()), i++) + dict->Set(field->name, String(row[i], row[i] + lengths[i])); + + return dict; +} + +void IdoMysqlConnection::DiscardRows(const IdoMysqlResult& result) +{ + Dictionary::Ptr row; + + while ((row = FetchRow(result))) + ; /* empty loop body */ +} + +void IdoMysqlConnection::ActivateObject(const DbObject::Ptr& dbobj) +{ + if (IsPaused()) + return; + +#ifdef I2_DEBUG /* I2_DEBUG */ + Log(LogDebug, "IdoMysqlConnection") + << "Scheduling object activation task for '" << dbobj->GetName1() << "!" << dbobj->GetName2() << "'."; +#endif /* I2_DEBUG */ + + m_QueryQueue.Enqueue([this, dbobj]() { InternalActivateObject(dbobj); }, PriorityNormal); +} + +void IdoMysqlConnection::InternalActivateObject(const DbObject::Ptr& dbobj) +{ + AssertOnWorkQueue(); + + if (IsPaused()) + return; + + if (!GetConnected()) + return; + + DbReference dbref = GetObjectID(dbobj); + std::ostringstream qbuf; + + if (!dbref.IsValid()) { + if (!dbobj->GetName2().IsEmpty()) { + qbuf << "INSERT INTO " + GetTablePrefix() + "objects (instance_id, objecttype_id, name1, name2, is_active) VALUES (" + << static_cast<long>(m_InstanceID) << ", " << dbobj->GetType()->GetTypeID() << ", " + << "'" << Escape(dbobj->GetName1()) << "', '" << Escape(dbobj->GetName2()) << "', 1)"; + } else { + qbuf << "INSERT INTO " + GetTablePrefix() + "objects (instance_id, objecttype_id, name1, is_active) VALUES (" + << static_cast<long>(m_InstanceID) << ", " << dbobj->GetType()->GetTypeID() << ", " + << "'" << Escape(dbobj->GetName1()) << "', 1)"; + } + + Query(qbuf.str()); + SetObjectID(dbobj, GetLastInsertID()); + } else { + qbuf << "UPDATE " + GetTablePrefix() + "objects SET is_active = 1 WHERE object_id = " << static_cast<long>(dbref); + IncreasePendingQueries(1); + AsyncQuery(qbuf.str()); + } +} + +void IdoMysqlConnection::DeactivateObject(const DbObject::Ptr& dbobj) +{ + if (IsPaused()) + return; + +#ifdef I2_DEBUG /* I2_DEBUG */ + Log(LogDebug, "IdoMysqlConnection") + << "Scheduling object deactivation task for '" << dbobj->GetName1() << "!" << dbobj->GetName2() << "'."; +#endif /* I2_DEBUG */ + + m_QueryQueue.Enqueue([this, dbobj]() { InternalDeactivateObject(dbobj); }, PriorityNormal); +} + +void IdoMysqlConnection::InternalDeactivateObject(const DbObject::Ptr& dbobj) +{ + AssertOnWorkQueue(); + + if (IsPaused()) + return; + + if (!GetConnected()) + return; + + DbReference dbref = GetObjectID(dbobj); + + if (!dbref.IsValid()) + return; + + std::ostringstream qbuf; + qbuf << "UPDATE " + GetTablePrefix() + "objects SET is_active = 0 WHERE object_id = " << static_cast<long>(dbref); + IncreasePendingQueries(1); + AsyncQuery(qbuf.str()); + + /* Note that we're _NOT_ clearing the db refs via SetReference/SetConfigUpdate/SetStatusUpdate + * because the object is still in the database. */ + + SetObjectActive(dbobj, false); +} + +bool IdoMysqlConnection::FieldToEscapedString(const String& key, const Value& value, Value *result) +{ + if (key == "instance_id") { + *result = static_cast<long>(m_InstanceID); + return true; + } else if (key == "session_token") { + *result = GetSessionToken(); + return true; + } + + Value rawvalue = DbValue::ExtractValue(value); + + if (rawvalue.GetType() == ValueEmpty) { + *result = "NULL"; + } else if (rawvalue.IsObjectType<ConfigObject>()) { + DbObject::Ptr dbobjcol = DbObject::GetOrCreateByObject(rawvalue); + + if (!dbobjcol) { + *result = 0; + return true; + } + + if (!IsIDCacheValid()) + return false; + + DbReference dbrefcol; + + if (DbValue::IsObjectInsertID(value)) { + dbrefcol = GetInsertID(dbobjcol); + + if (!dbrefcol.IsValid()) + return false; + } else { + dbrefcol = GetObjectID(dbobjcol); + + if (!dbrefcol.IsValid()) { + InternalActivateObject(dbobjcol); + + dbrefcol = GetObjectID(dbobjcol); + + if (!dbrefcol.IsValid()) + return false; + } + } + + *result = static_cast<long>(dbrefcol); + } else if (DbValue::IsTimestamp(value)) { + long ts = rawvalue; + std::ostringstream msgbuf; + msgbuf << "FROM_UNIXTIME(" << ts << ")"; + *result = Value(msgbuf.str()); + } else if (DbValue::IsObjectInsertID(value)) { + auto id = static_cast<long>(rawvalue); + + if (id <= 0) + return false; + + *result = id; + return true; + } else { + Value fvalue; + + if (rawvalue.IsBoolean()) + fvalue = Convert::ToLong(rawvalue); + else + fvalue = rawvalue; + + *result = "'" + Escape(fvalue) + "'"; + } + + return true; +} + +void IdoMysqlConnection::ExecuteQuery(const DbQuery& query) +{ + if (IsPaused() && GetPauseCalled()) + return; + + ASSERT(query.Category != DbCatInvalid); + +#ifdef I2_DEBUG /* I2_DEBUG */ + Log(LogDebug, "IdoMysqlConnection") + << "Scheduling execute query task, type " << query.Type << ", table '" << query.Table << "'."; +#endif /* I2_DEBUG */ + + IncreasePendingQueries(1); + m_QueryQueue.Enqueue([this, query]() { InternalExecuteQuery(query, -1); }, query.Priority, true); +} + +void IdoMysqlConnection::ExecuteMultipleQueries(const std::vector<DbQuery>& queries) +{ + if (IsPaused()) + return; + + if (queries.empty()) + return; + +#ifdef I2_DEBUG /* I2_DEBUG */ + Log(LogDebug, "IdoMysqlConnection") + << "Scheduling multiple execute query task, type " << queries[0].Type << ", table '" << queries[0].Table << "'."; +#endif /* I2_DEBUG */ + + IncreasePendingQueries(queries.size()); + m_QueryQueue.Enqueue([this, queries]() { InternalExecuteMultipleQueries(queries); }, queries[0].Priority, true); +} + +bool IdoMysqlConnection::CanExecuteQuery(const DbQuery& query) +{ + if (query.Object && !IsIDCacheValid()) + return false; + + if (query.WhereCriteria) { + ObjectLock olock(query.WhereCriteria); + Value value; + + for (const Dictionary::Pair& kv : query.WhereCriteria) { + if (!FieldToEscapedString(kv.first, kv.second, &value)) + return false; + } + } + + if (query.Fields) { + ObjectLock olock(query.Fields); + + for (const Dictionary::Pair& kv : query.Fields) { + Value value; + + if (!FieldToEscapedString(kv.first, kv.second, &value)) + return false; + } + } + + return true; +} + +void IdoMysqlConnection::InternalExecuteMultipleQueries(const std::vector<DbQuery>& queries) +{ + AssertOnWorkQueue(); + + if (IsPaused()) { + DecreasePendingQueries(queries.size()); + return; + } + + if (!GetConnected()) { + DecreasePendingQueries(queries.size()); + return; + } + + + for (const DbQuery& query : queries) { + ASSERT(query.Type == DbQueryNewTransaction || query.Category != DbCatInvalid); + + if (!CanExecuteQuery(query)) { + +#ifdef I2_DEBUG /* I2_DEBUG */ + Log(LogDebug, "IdoMysqlConnection") + << "Scheduling multiple execute query task again: Cannot execute query now. Type '" + << query.Type << "', table '" << query.Table << "', queue size: '" << GetPendingQueryCount() << "'."; +#endif /* I2_DEBUG */ + + m_QueryQueue.Enqueue([this, queries]() { InternalExecuteMultipleQueries(queries); }, query.Priority); + return; + } + } + + for (const DbQuery& query : queries) { + InternalExecuteQuery(query); + } +} + +void IdoMysqlConnection::InternalExecuteQuery(const DbQuery& query, int typeOverride) +{ + AssertOnWorkQueue(); + + if (IsPaused() && GetPauseCalled()) { + DecreasePendingQueries(1); + return; + } + + if (!GetConnected()) { + DecreasePendingQueries(1); + return; + } + + if (query.Type == DbQueryNewTransaction) { + DecreasePendingQueries(1); + InternalNewTransaction(); + return; + } + + /* check whether we're allowed to execute the query first */ + if (GetCategoryFilter() != DbCatEverything && (query.Category & GetCategoryFilter()) == 0) { + DecreasePendingQueries(1); + return; + } + + if (query.Object && query.Object->GetObject()->GetExtension("agent_check").ToBool()) { + DecreasePendingQueries(1); + return; + } + + /* check if there are missing object/insert ids and re-enqueue the query */ + if (!CanExecuteQuery(query)) { + +#ifdef I2_DEBUG /* I2_DEBUG */ + Log(LogDebug, "IdoMysqlConnection") + << "Scheduling execute query task again: Cannot execute query now. Type '" + << typeOverride << "', table '" << query.Table << "', queue size: '" << GetPendingQueryCount() << "'."; +#endif /* I2_DEBUG */ + + m_QueryQueue.Enqueue([this, query, typeOverride]() { InternalExecuteQuery(query, typeOverride); }, query.Priority); + return; + } + + std::ostringstream qbuf, where; + int type; + + if (query.WhereCriteria) { + where << " WHERE "; + + ObjectLock olock(query.WhereCriteria); + Value value; + bool first = true; + + for (const Dictionary::Pair& kv : query.WhereCriteria) { + if (!FieldToEscapedString(kv.first, kv.second, &value)) { + +#ifdef I2_DEBUG /* I2_DEBUG */ + Log(LogDebug, "IdoMysqlConnection") + << "Scheduling execute query task again: Cannot execute query now. Type '" + << typeOverride << "', table '" << query.Table << "', queue size: '" << GetPendingQueryCount() << "'."; +#endif /* I2_DEBUG */ + + m_QueryQueue.Enqueue([this, query]() { InternalExecuteQuery(query, -1); }, query.Priority); + return; + } + + if (!first) + where << " AND "; + + where << kv.first << " = " << value; + + if (first) + first = false; + } + } + + type = (typeOverride != -1) ? typeOverride : query.Type; + + bool upsert = false; + + if ((type & DbQueryInsert) && (type & DbQueryUpdate)) { + bool hasid = false; + + if (query.Object) { + if (query.ConfigUpdate) + hasid = GetConfigUpdate(query.Object); + else if (query.StatusUpdate) + hasid = GetStatusUpdate(query.Object); + } + + if (!hasid) + upsert = true; + + type = DbQueryUpdate; + } + + if ((type & DbQueryInsert) && (type & DbQueryDelete)) { + std::ostringstream qdel; + qdel << "DELETE FROM " << GetTablePrefix() << query.Table << where.str(); + IncreasePendingQueries(1); + AsyncQuery(qdel.str()); + + type = DbQueryInsert; + } + + switch (type) { + case DbQueryInsert: + qbuf << "INSERT INTO " << GetTablePrefix() << query.Table; + break; + case DbQueryUpdate: + qbuf << "UPDATE " << GetTablePrefix() << query.Table << " SET"; + break; + case DbQueryDelete: + qbuf << "DELETE FROM " << GetTablePrefix() << query.Table; + break; + default: + VERIFY(!"Invalid query type."); + } + + if (type == DbQueryInsert || type == DbQueryUpdate) { + std::ostringstream colbuf, valbuf; + + if (type == DbQueryUpdate && query.Fields->GetLength() == 0) + return; + + ObjectLock olock(query.Fields); + + bool first = true; + for (const Dictionary::Pair& kv : query.Fields) { + Value value; + + if (!FieldToEscapedString(kv.first, kv.second, &value)) { + +#ifdef I2_DEBUG /* I2_DEBUG */ + Log(LogDebug, "IdoMysqlConnection") + << "Scheduling execute query task again: Cannot extract required INSERT/UPDATE fields, key '" + << kv.first << "', val '" << kv.second << "', type " << typeOverride << ", table '" << query.Table << "'."; +#endif /* I2_DEBUG */ + + m_QueryQueue.Enqueue([this, query]() { InternalExecuteQuery(query, -1); }, query.Priority); + return; + } + + if (type == DbQueryInsert) { + if (!first) { + colbuf << ", "; + valbuf << ", "; + } + + colbuf << kv.first; + valbuf << value; + } else { + if (!first) + qbuf << ", "; + + qbuf << " " << kv.first << " = " << value; + } + + if (first) + first = false; + } + + if (type == DbQueryInsert) + qbuf << " (" << colbuf.str() << ") VALUES (" << valbuf.str() << ")"; + } + + if (type != DbQueryInsert) + qbuf << where.str(); + + AsyncQuery(qbuf.str(), [this, query, type, upsert](const IdoMysqlResult&) { FinishExecuteQuery(query, type, upsert); }); +} + +void IdoMysqlConnection::FinishExecuteQuery(const DbQuery& query, int type, bool upsert) +{ + if (upsert && GetAffectedRows() == 0) { + +#ifdef I2_DEBUG /* I2_DEBUG */ + Log(LogDebug, "IdoMysqlConnection") + << "Rescheduling DELETE/INSERT query: Upsert UPDATE did not affect rows, type " << type << ", table '" << query.Table << "'."; +#endif /* I2_DEBUG */ + + IncreasePendingQueries(1); + m_QueryQueue.Enqueue([this, query]() { InternalExecuteQuery(query, DbQueryDelete | DbQueryInsert); }, query.Priority); + + return; + } + + if (type == DbQueryInsert && query.Object) { + if (query.ConfigUpdate) { + SetInsertID(query.Object, GetLastInsertID()); + SetConfigUpdate(query.Object, true); + } else if (query.StatusUpdate) + SetStatusUpdate(query.Object, true); + } + + if (type == DbQueryInsert && query.Table == "notifications" && query.NotificationInsertID) + query.NotificationInsertID->SetValue(static_cast<long>(GetLastInsertID())); +} + +void IdoMysqlConnection::CleanUpExecuteQuery(const String& table, const String& time_column, double max_age) +{ + if (IsPaused()) + return; + +#ifdef I2_DEBUG /* I2_DEBUG */ + Log(LogDebug, "IdoMysqlConnection") + << "Rescheduling cleanup query for table '" << table << "' and column '" + << time_column << "'. max_age is set to '" << max_age << "'."; +#endif /* I2_DEBUG */ + + IncreasePendingQueries(1); + m_QueryQueue.Enqueue([this, table, time_column, max_age]() { InternalCleanUpExecuteQuery(table, time_column, max_age); }, PriorityLow, true); +} + +void IdoMysqlConnection::InternalCleanUpExecuteQuery(const String& table, const String& time_column, double max_age) +{ + AssertOnWorkQueue(); + + if (IsPaused()) { + DecreasePendingQueries(1); + return; + } + + if (!GetConnected()) { + DecreasePendingQueries(1); + return; + } + + AsyncQuery("DELETE FROM " + GetTablePrefix() + table + " WHERE instance_id = " + + Convert::ToString(static_cast<long>(m_InstanceID)) + " AND " + time_column + + " < FROM_UNIXTIME(" + Convert::ToString(static_cast<long>(max_age)) + ")"); +} + +void IdoMysqlConnection::FillIDCache(const DbType::Ptr& type) +{ + String query = "SELECT " + type->GetIDColumn() + " AS object_id, " + type->GetTable() + "_id, config_hash FROM " + GetTablePrefix() + type->GetTable() + "s"; + IdoMysqlResult result = Query(query); + + Dictionary::Ptr row; + + while ((row = FetchRow(result))) { + DbReference dbref(row->Get("object_id")); + SetInsertID(type, dbref, DbReference(row->Get(type->GetTable() + "_id"))); + SetConfigHash(type, dbref, row->Get("config_hash")); + } +} + +int IdoMysqlConnection::GetPendingQueryCount() const +{ + return m_QueryQueue.GetLength(); +} |