1 /* Icinga 2 | (c) 2012 Icinga GmbH | GPLv2+ */
3 #include "db_ido_mysql/idomysqlconnection.hpp"
4 #include "db_ido_mysql/idomysqlconnection-ti.cpp"
5 #include "db_ido/dbtype.hpp"
6 #include "db_ido/dbvalue.hpp"
7 #include "base/logger.hpp"
8 #include "base/objectlock.hpp"
9 #include "base/convert.hpp"
10 #include "base/utility.hpp"
11 #include "base/perfdatavalue.hpp"
12 #include "base/application.hpp"
13 #include "base/configtype.hpp"
14 #include "base/exception.hpp"
15 #include "base/statsfunction.hpp"
16 #include <boost/tuple/tuple.hpp>
19 using namespace icinga;
21 REGISTER_TYPE(IdoMysqlConnection);
22 REGISTER_STATSFUNCTION(IdoMysqlConnection, &IdoMysqlConnection::StatsFunc);
24 void IdoMysqlConnection::OnConfigLoaded()
26 ObjectImpl<IdoMysqlConnection>::OnConfigLoaded();
28 m_QueryQueue.SetName("IdoMysqlConnection, " + GetName());
30 Library shimLibrary{"mysql_shim"};
32 auto create_mysql_shim = shimLibrary.GetSymbolAddress<create_mysql_shim_ptr>("create_mysql_shim");
34 m_Mysql.reset(create_mysql_shim());
36 std::swap(m_Library, shimLibrary);
39 void IdoMysqlConnection::StatsFunc(const Dictionary::Ptr& status, const Array::Ptr& perfdata)
43 for (const IdoMysqlConnection::Ptr& idomysqlconnection : ConfigType::GetObjectsByType<IdoMysqlConnection>()) {
44 size_t queryQueueItems = idomysqlconnection->m_QueryQueue.GetLength();
45 double queryQueueItemRate = idomysqlconnection->m_QueryQueue.GetTaskCount(60) / 60.0;
47 nodes.emplace_back(idomysqlconnection->GetName(), new Dictionary({
48 { "version", idomysqlconnection->GetSchemaVersion() },
49 { "instance_name", idomysqlconnection->GetInstanceName() },
50 { "connected", idomysqlconnection->GetConnected() },
51 { "query_queue_items", queryQueueItems },
52 { "query_queue_item_rate", queryQueueItemRate }
55 perfdata->Add(new PerfdataValue("idomysqlconnection_" + idomysqlconnection->GetName() + "_queries_rate", idomysqlconnection->GetQueryCount(60) / 60.0));
56 perfdata->Add(new PerfdataValue("idomysqlconnection_" + idomysqlconnection->GetName() + "_queries_1min", idomysqlconnection->GetQueryCount(60)));
57 perfdata->Add(new PerfdataValue("idomysqlconnection_" + idomysqlconnection->GetName() + "_queries_5mins", idomysqlconnection->GetQueryCount(5 * 60)));
58 perfdata->Add(new PerfdataValue("idomysqlconnection_" + idomysqlconnection->GetName() + "_queries_15mins", idomysqlconnection->GetQueryCount(15 * 60)));
59 perfdata->Add(new PerfdataValue("idomysqlconnection_" + idomysqlconnection->GetName() + "_query_queue_items", queryQueueItems));
60 perfdata->Add(new PerfdataValue("idomysqlconnection_" + idomysqlconnection->GetName() + "_query_queue_item_rate", queryQueueItemRate));
63 status->Set("idomysqlconnection", new Dictionary(std::move(nodes)));
66 void IdoMysqlConnection::Resume()
68 Log(LogInformation, "IdoMysqlConnection")
69 << "'" << GetName() << "' resumed.";
73 m_QueryQueue.SetExceptionCallback(std::bind(&IdoMysqlConnection::ExceptionHandler, this, _1));
75 /* Immediately try to connect on Resume() without timer. */
76 m_QueryQueue.Enqueue(std::bind(&IdoMysqlConnection::Reconnect, this), PriorityImmediate);
78 m_TxTimer = new Timer();
79 m_TxTimer->SetInterval(1);
80 m_TxTimer->OnTimerExpired.connect(std::bind(&IdoMysqlConnection::TxTimerHandler, this));
83 m_ReconnectTimer = new Timer();
84 m_ReconnectTimer->SetInterval(10);
85 m_ReconnectTimer->OnTimerExpired.connect(std::bind(&IdoMysqlConnection::ReconnectTimerHandler, this));
86 m_ReconnectTimer->Start();
88 /* Start with queries after connect. */
89 DbConnection::Resume();
91 ASSERT(m_Mysql->thread_safe());
94 void IdoMysqlConnection::Pause()
96 Log(LogDebug, "IdoMysqlConnection")
97 << "Attempting to pause '" << GetName() << "'.";
99 DbConnection::Pause();
101 m_ReconnectTimer.reset();
103 #ifdef I2_DEBUG /* I2_DEBUG */
104 Log(LogDebug, "IdoMysqlConnection")
105 << "Rescheduling disconnect task.";
106 #endif /* I2_DEBUG */
108 m_QueryQueue.Enqueue(std::bind(&IdoMysqlConnection::Disconnect, this), PriorityLow);
110 /* Work on remaining tasks but never delete the threads, for HA resuming later. */
113 Log(LogInformation, "IdoMysqlConnection")
114 << "'" << GetName() << "' paused.";
118 void IdoMysqlConnection::ExceptionHandler(boost::exception_ptr exp)
120 Log(LogCritical, "IdoMysqlConnection", "Exception during database operation: Verify that your database is operational!");
122 Log(LogDebug, "IdoMysqlConnection")
123 << "Exception during database operation: " << DiagnosticInformation(std::move(exp));
125 if (GetConnected()) {
126 m_Mysql->close(&m_Connection);
132 void IdoMysqlConnection::AssertOnWorkQueue()
134 ASSERT(m_QueryQueue.IsWorkerThread());
137 void IdoMysqlConnection::Disconnect()
145 m_Mysql->close(&m_Connection);
149 Log(LogInformation, "IdoMysqlConnection")
150 << "Disconnected from '" << GetName() << "' database '" << GetDatabase() << "'.";
153 void IdoMysqlConnection::TxTimerHandler()
158 void IdoMysqlConnection::NewTransaction()
163 #ifdef I2_DEBUG /* I2_DEBUG */
164 Log(LogDebug, "IdoMysqlConnection")
165 << "Scheduling new transaction and finishing async queries.";
166 #endif /* I2_DEBUG */
168 m_QueryQueue.Enqueue(std::bind(&IdoMysqlConnection::InternalNewTransaction, this), PriorityNormal);
169 m_QueryQueue.Enqueue(std::bind(&IdoMysqlConnection::FinishAsyncQueries, this), PriorityNormal);
172 void IdoMysqlConnection::InternalNewTransaction()
179 AsyncQuery("COMMIT");
183 void IdoMysqlConnection::ReconnectTimerHandler()
185 #ifdef I2_DEBUG /* I2_DEBUG */
186 Log(LogDebug, "IdoMysqlConnection")
187 << "Scheduling reconnect task.";
188 #endif /* I2_DEBUG */
190 /* Only allow Reconnect events with high priority. */
191 m_QueryQueue.Enqueue(std::bind(&IdoMysqlConnection::Reconnect, this), PriorityImmediate);
194 void IdoMysqlConnection::Reconnect()
201 CONTEXT("Reconnecting to MySQL IDO database '" + GetName() + "'");
203 double startTime = Utility::GetTime();
205 SetShouldConnect(true);
207 bool reconnect = false;
209 /* Ensure to close old connections first. */
210 if (GetConnected()) {
211 /* Check if we're really still connected */
212 if (m_Mysql->ping(&m_Connection) == 0)
215 m_Mysql->close(&m_Connection);
220 Log(LogDebug, "IdoMysqlConnection")
221 << "Reconnect: Clearing ID cache.";
225 String ihost, isocket_path, iuser, ipasswd, idb;
226 String isslKey, isslCert, isslCa, isslCaPath, isslCipher;
227 const char *host, *socket_path, *user , *passwd, *db;
228 const char *sslKey, *sslCert, *sslCa, *sslCaPath, *sslCipher;
233 isocket_path = GetSocketPath();
235 ipasswd = GetPassword();
238 enableSsl = GetEnableSsl();
239 isslKey = GetSslKey();
240 isslCert = GetSslCert();
242 isslCaPath = GetSslCapath();
243 isslCipher = GetSslCipher();
245 host = (!ihost.IsEmpty()) ? ihost.CStr() : nullptr;
247 socket_path = (!isocket_path.IsEmpty()) ? isocket_path.CStr() : nullptr;
248 user = (!iuser.IsEmpty()) ? iuser.CStr() : nullptr;
249 passwd = (!ipasswd.IsEmpty()) ? ipasswd.CStr() : nullptr;
250 db = (!idb.IsEmpty()) ? idb.CStr() : nullptr;
252 sslKey = (!isslKey.IsEmpty()) ? isslKey.CStr() : nullptr;
253 sslCert = (!isslCert.IsEmpty()) ? isslCert.CStr() : nullptr;
254 sslCa = (!isslCa.IsEmpty()) ? isslCa.CStr() : nullptr;
255 sslCaPath = (!isslCaPath.IsEmpty()) ? isslCaPath.CStr() : nullptr;
256 sslCipher = (!isslCipher.IsEmpty()) ? isslCipher.CStr() : nullptr;
259 if (!m_Mysql->init(&m_Connection)) {
260 Log(LogCritical, "IdoMysqlConnection")
261 << "mysql_init() failed: out of memory";
263 BOOST_THROW_EXCEPTION(std::bad_alloc());
267 m_Mysql->ssl_set(&m_Connection, sslKey, sslCert, sslCa, sslCaPath, sslCipher);
269 if (!m_Mysql->real_connect(&m_Connection, host, user, passwd, db, port, socket_path, CLIENT_FOUND_ROWS | CLIENT_MULTI_STATEMENTS)) {
270 Log(LogCritical, "IdoMysqlConnection")
271 << "Connection to database '" << db << "' with user '" << user << "' on '" << host << ":" << port
272 << "' " << (enableSsl ? "(SSL enabled) " : "") << "failed: \"" << m_Mysql->error(&m_Connection) << "\"";
274 BOOST_THROW_EXCEPTION(std::runtime_error(m_Mysql->error(&m_Connection)));
277 Log(LogNotice, "IdoMysqlConnection")
278 << "Reconnect: '" << GetName() << "' is now connected to database '" << GetDatabase() << "'.";
282 IdoMysqlResult result = Query("SELECT @@global.max_allowed_packet AS max_allowed_packet");
284 Dictionary::Ptr row = FetchRow(result);
287 m_MaxPacketSize = row->Get("max_allowed_packet");
289 m_MaxPacketSize = 64 * 1024;
293 String dbVersionName = "idoutils";
294 result = Query("SELECT version FROM " + GetTablePrefix() + "dbversion WHERE name='" + Escape(dbVersionName) + "'");
296 row = FetchRow(result);
299 m_Mysql->close(&m_Connection);
302 Log(LogCritical, "IdoMysqlConnection", "Schema does not provide any valid version! Verify your schema installation.");
304 BOOST_THROW_EXCEPTION(std::runtime_error("Invalid schema."));
309 String version = row->Get("version");
311 SetSchemaVersion(version);
313 if (Utility::CompareVersion(IDO_COMPAT_SCHEMA_VERSION, version) < 0) {
314 m_Mysql->close(&m_Connection);
317 Log(LogCritical, "IdoMysqlConnection")
318 << "Schema version '" << version << "' does not match the required version '"
319 << IDO_COMPAT_SCHEMA_VERSION << "' (or newer)! Please check the upgrade documentation at "
320 << "https://docs.icinga.com/icinga2/latest/doc/module/icinga2/chapter/upgrading-icinga-2#upgrading-mysql-db";
322 BOOST_THROW_EXCEPTION(std::runtime_error("Schema version mismatch."));
325 String instanceName = GetInstanceName();
327 result = Query("SELECT instance_id FROM " + GetTablePrefix() + "instances WHERE instance_name = '" + Escape(instanceName) + "'");
328 row = FetchRow(result);
331 Query("INSERT INTO " + GetTablePrefix() + "instances (instance_name, instance_description) VALUES ('" + Escape(instanceName) + "', '" + Escape(GetInstanceDescription()) + "')");
332 m_InstanceID = GetLastInsertID();
334 m_InstanceID = DbReference(row->Get("instance_id"));
339 Endpoint::Ptr my_endpoint = Endpoint::GetLocalEndpoint();
341 /* we have an endpoint in a cluster setup, so decide if we can proceed here */
342 if (my_endpoint && GetHAMode() == HARunOnce) {
343 /* get the current endpoint writing to programstatus table */
344 result = Query("SELECT UNIX_TIMESTAMP(status_update_time) AS status_update_time, endpoint_name FROM " +
345 GetTablePrefix() + "programstatus WHERE instance_id = " + Convert::ToString(m_InstanceID));
346 row = FetchRow(result);
349 String endpoint_name;
352 endpoint_name = row->Get("endpoint_name");
354 Log(LogNotice, "IdoMysqlConnection", "Empty program status table");
356 /* if we did not write into the database earlier, another instance is active */
357 if (endpoint_name != my_endpoint->GetName()) {
358 double status_update_time;
361 status_update_time = row->Get("status_update_time");
363 status_update_time = 0;
365 double now = Utility::GetTime();
367 double status_update_age = now - status_update_time;
368 double failoverTimeout = GetFailoverTimeout();
370 if (status_update_age < failoverTimeout) {
371 Log(LogInformation, "IdoMysqlConnection")
372 << "Last update by endpoint '" << endpoint_name << "' was "
373 << status_update_age << "s ago (< failover timeout of " << failoverTimeout << "s). Retrying.";
375 m_Mysql->close(&m_Connection);
377 SetShouldConnect(false);
382 /* activate the IDO only, if we're authoritative in this zone */
384 Log(LogNotice, "IdoMysqlConnection")
385 << "Local endpoint '" << my_endpoint->GetName() << "' is not authoritative, bailing out.";
387 m_Mysql->close(&m_Connection);
393 SetLastFailover(now);
395 Log(LogInformation, "IdoMysqlConnection")
396 << "Last update by endpoint '" << endpoint_name << "' was "
397 << status_update_age << "s ago. Taking over '" << GetName() << "' in HA zone '" << Zone::GetLocalZone()->GetName() << "'.";
400 Log(LogNotice, "IdoMysqlConnection", "Enabling IDO connection in HA zone.");
403 Log(LogInformation, "IdoMysqlConnection")
404 << "MySQL IDO instance id: " << static_cast<long>(m_InstanceID) << " (schema version: '" + version + "')";
406 /* set session time zone to utc */
407 Query("SET SESSION TIME_ZONE='+00:00'");
409 Query("SET SESSION SQL_MODE='NO_AUTO_VALUE_ON_ZERO'");
413 /* update programstatus table */
414 UpdateProgramStatus();
416 /* record connection */
417 Query("INSERT INTO " + GetTablePrefix() + "conninfo " +
418 "(instance_id, connect_time, last_checkin_time, agent_name, agent_version, connect_type, data_start_time) VALUES ("
419 + Convert::ToString(static_cast<long>(m_InstanceID)) + ", NOW(), NOW(), 'icinga2 db_ido_mysql', '" + Escape(Application::GetAppVersion())
420 + "', '" + (reconnect ? "RECONNECT" : "INITIAL") + "', NOW())");
422 /* clear config tables for the initial config dump */
425 std::ostringstream q1buf;
426 q1buf << "SELECT object_id, objecttype_id, name1, name2, is_active FROM " + GetTablePrefix() + "objects WHERE instance_id = " << static_cast<long>(m_InstanceID);
427 result = Query(q1buf.str());
429 std::vector<DbObject::Ptr> activeDbObjs;
431 while ((row = FetchRow(result))) {
432 DbType::Ptr dbtype = DbType::GetByID(row->Get("objecttype_id"));
437 DbObject::Ptr dbobj = dbtype->GetOrCreateObjectByName(row->Get("name1"), row->Get("name2"));
438 SetObjectID(dbobj, DbReference(row->Get("object_id")));
439 bool active = row->Get("is_active");
440 SetObjectActive(dbobj, active);
443 activeDbObjs.emplace_back(std::move(dbobj));
446 SetIDCacheValid(true);
448 EnableActiveChangedHandler();
450 for (const DbObject::Ptr& dbobj : activeDbObjs) {
451 if (dbobj->GetObject())
454 Log(LogNotice, "IdoMysqlConnection")
455 << "Deactivate deleted object name1: '" << dbobj->GetName1()
456 << "' name2: '" << dbobj->GetName2() + "'.";
457 DeactivateObject(dbobj);
462 #ifdef I2_DEBUG /* I2_DEBUG */
463 Log(LogDebug, "IdoMysqlConnection")
464 << "Scheduling session table clear and finish connect task.";
465 #endif /* I2_DEBUG */
467 m_QueryQueue.Enqueue(std::bind(&IdoMysqlConnection::ClearTablesBySession, this), PriorityNormal);
469 m_QueryQueue.Enqueue(std::bind(&IdoMysqlConnection::FinishConnect, this, startTime), PriorityNormal);
472 void IdoMysqlConnection::FinishConnect(double startTime)
479 FinishAsyncQueries();
481 Log(LogInformation, "IdoMysqlConnection")
482 << "Finished reconnecting to '" << GetName() << "' database '" << GetDatabase() << "' in "
483 << std::setw(2) << Utility::GetTime() - startTime << " second(s).";
489 void IdoMysqlConnection::ClearTablesBySession()
491 /* delete all comments and downtimes without current session token */
492 ClearTableBySession("comments");
493 ClearTableBySession("scheduleddowntime");
496 void IdoMysqlConnection::ClearTableBySession(const String& table)
498 Query("DELETE FROM " + GetTablePrefix() + table + " WHERE instance_id = " +
499 Convert::ToString(static_cast<long>(m_InstanceID)) + " AND session_token <> " +
500 Convert::ToString(GetSessionToken()));
503 void IdoMysqlConnection::AsyncQuery(const String& query, const std::function<void (const IdoMysqlResult&)>& callback)
509 /* XXX: Important: The callback must not immediately execute a query, but enqueue it!
510 * See https://github.com/Icinga/icinga2/issues/4603 for details.
512 aq.Callback = callback;
513 m_AsyncQueries.emplace_back(std::move(aq));
515 if (m_AsyncQueries.size() > 25000) {
516 FinishAsyncQueries();
517 InternalNewTransaction();
521 void IdoMysqlConnection::FinishAsyncQueries()
523 std::vector<IdoAsyncQuery> queries;
524 m_AsyncQueries.swap(queries);
526 std::vector<IdoAsyncQuery>::size_type offset = 0;
528 while (offset < queries.size()) {
529 std::ostringstream querybuf;
531 std::vector<IdoAsyncQuery>::size_type count = 0;
532 size_t num_bytes = 0;
534 for (std::vector<IdoAsyncQuery>::size_type i = offset; i < queries.size(); i++) {
535 const IdoAsyncQuery& aq = queries[i];
537 size_t size_query = aq.Query.GetLength() + 1;
540 if (num_bytes + size_query > m_MaxPacketSize - 512)
546 IncreaseQueryCount();
549 Log(LogDebug, "IdoMysqlConnection")
550 << "Query: " << aq.Query;
552 querybuf << aq.Query;
553 num_bytes += size_query;
556 String query = querybuf.str();
558 if (m_Mysql->query(&m_Connection, query.CStr()) != 0) {
559 std::ostringstream msgbuf;
560 String message = m_Mysql->error(&m_Connection);
561 msgbuf << "Error \"" << message << "\" when executing query \"" << query << "\"";
562 Log(LogCritical, "IdoMysqlConnection", msgbuf.str());
564 BOOST_THROW_EXCEPTION(
566 << errinfo_message(m_Mysql->error(&m_Connection))
567 << errinfo_database_query(query)
571 for (std::vector<IdoAsyncQuery>::size_type i = offset; i < offset + count; i++) {
572 const IdoAsyncQuery& aq = queries[i];
574 MYSQL_RES *result = m_Mysql->store_result(&m_Connection);
576 m_AffectedRows = m_Mysql->affected_rows(&m_Connection);
578 IdoMysqlResult iresult;
581 if (m_Mysql->field_count(&m_Connection) > 0) {
582 std::ostringstream msgbuf;
583 String message = m_Mysql->error(&m_Connection);
584 msgbuf << "Error \"" << message << "\" when executing query \"" << aq.Query << "\"";
585 Log(LogCritical, "IdoMysqlConnection", msgbuf.str());
587 BOOST_THROW_EXCEPTION(
589 << errinfo_message(m_Mysql->error(&m_Connection))
590 << errinfo_database_query(query)
594 iresult = IdoMysqlResult(result, std::bind(&MysqlInterface::free_result, std::cref(m_Mysql), _1));
597 aq.Callback(iresult);
599 if (m_Mysql->next_result(&m_Connection) > 0) {
600 std::ostringstream msgbuf;
601 String message = m_Mysql->error(&m_Connection);
602 msgbuf << "Error \"" << message << "\" when executing query \"" << query << "\"";
603 Log(LogCritical, "IdoMysqlConnection", msgbuf.str());
605 BOOST_THROW_EXCEPTION(
607 << errinfo_message(m_Mysql->error(&m_Connection))
608 << errinfo_database_query(query)
617 IdoMysqlResult IdoMysqlConnection::Query(const String& query)
621 /* finish all async queries to maintain the right order for queries */
622 FinishAsyncQueries();
624 Log(LogDebug, "IdoMysqlConnection")
625 << "Query: " << query;
627 IncreaseQueryCount();
629 if (m_Mysql->query(&m_Connection, query.CStr()) != 0) {
630 std::ostringstream msgbuf;
631 String message = m_Mysql->error(&m_Connection);
632 msgbuf << "Error \"" << message << "\" when executing query \"" << query << "\"";
633 Log(LogCritical, "IdoMysqlConnection", msgbuf.str());
635 BOOST_THROW_EXCEPTION(
637 << errinfo_message(m_Mysql->error(&m_Connection))
638 << errinfo_database_query(query)
642 MYSQL_RES *result = m_Mysql->store_result(&m_Connection);
644 m_AffectedRows = m_Mysql->affected_rows(&m_Connection);
647 if (m_Mysql->field_count(&m_Connection) > 0) {
648 std::ostringstream msgbuf;
649 String message = m_Mysql->error(&m_Connection);
650 msgbuf << "Error \"" << message << "\" when executing query \"" << query << "\"";
651 Log(LogCritical, "IdoMysqlConnection", msgbuf.str());
653 BOOST_THROW_EXCEPTION(
655 << errinfo_message(m_Mysql->error(&m_Connection))
656 << errinfo_database_query(query)
660 return IdoMysqlResult();
663 return IdoMysqlResult(result, std::bind(&MysqlInterface::free_result, std::cref(m_Mysql), _1));
666 DbReference IdoMysqlConnection::GetLastInsertID()
670 return {static_cast<long>(m_Mysql->insert_id(&m_Connection))};
673 int IdoMysqlConnection::GetAffectedRows()
677 return m_AffectedRows;
680 String IdoMysqlConnection::Escape(const String& s)
684 String utf8s = Utility::ValidateUTF8(s);
686 size_t length = utf8s.GetLength();
687 auto *to = new char[utf8s.GetLength() * 2 + 1];
689 m_Mysql->real_escape_string(&m_Connection, to, utf8s.CStr(), length);
691 String result = String(to);
698 Dictionary::Ptr IdoMysqlConnection::FetchRow(const IdoMysqlResult& result)
704 unsigned long *lengths, i;
706 row = m_Mysql->fetch_row(result.get());
711 lengths = m_Mysql->fetch_lengths(result.get());
716 Dictionary::Ptr dict = new Dictionary();
718 m_Mysql->field_seek(result.get(), 0);
719 for (field = m_Mysql->fetch_field(result.get()), i = 0; field; field = m_Mysql->fetch_field(result.get()), i++)
720 dict->Set(field->name, String(row[i], row[i] + lengths[i]));
725 void IdoMysqlConnection::DiscardRows(const IdoMysqlResult& result)
729 while ((row = FetchRow(result)))
730 ; /* empty loop body */
733 void IdoMysqlConnection::ActivateObject(const DbObject::Ptr& dbobj)
738 #ifdef I2_DEBUG /* I2_DEBUG */
739 Log(LogDebug, "IdoMysqlConnection")
740 << "Scheduling object activation task for '" << dbobj->GetName1() << "!" << dbobj->GetName2() << "'.";
741 #endif /* I2_DEBUG */
743 m_QueryQueue.Enqueue(std::bind(&IdoMysqlConnection::InternalActivateObject, this, dbobj), PriorityNormal);
746 void IdoMysqlConnection::InternalActivateObject(const DbObject::Ptr& dbobj)
756 DbReference dbref = GetObjectID(dbobj);
757 std::ostringstream qbuf;
759 if (!dbref.IsValid()) {
760 if (!dbobj->GetName2().IsEmpty()) {
761 qbuf << "INSERT INTO " + GetTablePrefix() + "objects (instance_id, objecttype_id, name1, name2, is_active) VALUES ("
762 << static_cast<long>(m_InstanceID) << ", " << dbobj->GetType()->GetTypeID() << ", "
763 << "'" << Escape(dbobj->GetName1()) << "', '" << Escape(dbobj->GetName2()) << "', 1)";
765 qbuf << "INSERT INTO " + GetTablePrefix() + "objects (instance_id, objecttype_id, name1, is_active) VALUES ("
766 << static_cast<long>(m_InstanceID) << ", " << dbobj->GetType()->GetTypeID() << ", "
767 << "'" << Escape(dbobj->GetName1()) << "', 1)";
771 SetObjectID(dbobj, GetLastInsertID());
773 qbuf << "UPDATE " + GetTablePrefix() + "objects SET is_active = 1 WHERE object_id = " << static_cast<long>(dbref);
774 AsyncQuery(qbuf.str());
778 void IdoMysqlConnection::DeactivateObject(const DbObject::Ptr& dbobj)
783 #ifdef I2_DEBUG /* I2_DEBUG */
784 Log(LogDebug, "IdoMysqlConnection")
785 << "Scheduling object deactivation task for '" << dbobj->GetName1() << "!" << dbobj->GetName2() << "'.";
786 #endif /* I2_DEBUG */
788 m_QueryQueue.Enqueue(std::bind(&IdoMysqlConnection::InternalDeactivateObject, this, dbobj), PriorityNormal);
791 void IdoMysqlConnection::InternalDeactivateObject(const DbObject::Ptr& dbobj)
801 DbReference dbref = GetObjectID(dbobj);
803 if (!dbref.IsValid())
806 std::ostringstream qbuf;
807 qbuf << "UPDATE " + GetTablePrefix() + "objects SET is_active = 0 WHERE object_id = " << static_cast<long>(dbref);
808 AsyncQuery(qbuf.str());
810 /* Note that we're _NOT_ clearing the db refs via SetReference/SetConfigUpdate/SetStatusUpdate
811 * because the object is still in the database. */
814 bool IdoMysqlConnection::FieldToEscapedString(const String& key, const Value& value, Value *result)
816 if (key == "instance_id") {
817 *result = static_cast<long>(m_InstanceID);
819 } else if (key == "session_token") {
820 *result = GetSessionToken();
824 Value rawvalue = DbValue::ExtractValue(value);
826 if (rawvalue.IsObjectType<ConfigObject>()) {
827 DbObject::Ptr dbobjcol = DbObject::GetOrCreateByObject(rawvalue);
834 if (!IsIDCacheValid())
837 DbReference dbrefcol;
839 if (DbValue::IsObjectInsertID(value)) {
840 dbrefcol = GetInsertID(dbobjcol);
842 if (!dbrefcol.IsValid())
845 dbrefcol = GetObjectID(dbobjcol);
847 if (!dbrefcol.IsValid()) {
848 InternalActivateObject(dbobjcol);
850 dbrefcol = GetObjectID(dbobjcol);
852 if (!dbrefcol.IsValid())
857 *result = static_cast<long>(dbrefcol);
858 } else if (DbValue::IsTimestamp(value)) {
860 std::ostringstream msgbuf;
861 msgbuf << "FROM_UNIXTIME(" << ts << ")";
862 *result = Value(msgbuf.str());
863 } else if (DbValue::IsObjectInsertID(value)) {
864 auto id = static_cast<long>(rawvalue);
874 if (rawvalue.IsBoolean())
875 fvalue = Convert::ToLong(rawvalue);
879 *result = "'" + Escape(fvalue) + "'";
885 void IdoMysqlConnection::ExecuteQuery(const DbQuery& query)
890 ASSERT(query.Category != DbCatInvalid);
892 #ifdef I2_DEBUG /* I2_DEBUG */
893 Log(LogDebug, "IdoMysqlConnection")
894 << "Scheduling execute query task, type " << query.Type << ", table '" << query.Table << "'.";
895 #endif /* I2_DEBUG */
897 m_QueryQueue.Enqueue(std::bind(&IdoMysqlConnection::InternalExecuteQuery, this, query, -1), query.Priority, true);
900 void IdoMysqlConnection::ExecuteMultipleQueries(const std::vector<DbQuery>& queries)
908 #ifdef I2_DEBUG /* I2_DEBUG */
909 Log(LogDebug, "IdoMysqlConnection")
910 << "Scheduling multiple execute query task, type " << queries[0].Type << ", table '" << queries[0].Table << "'.";
911 #endif /* I2_DEBUG */
913 m_QueryQueue.Enqueue(std::bind(&IdoMysqlConnection::InternalExecuteMultipleQueries, this, queries), queries[0].Priority, true);
916 bool IdoMysqlConnection::CanExecuteQuery(const DbQuery& query)
918 if (query.Object && !IsIDCacheValid())
921 if (query.WhereCriteria) {
922 ObjectLock olock(query.WhereCriteria);
925 for (const Dictionary::Pair& kv : query.WhereCriteria) {
926 if (!FieldToEscapedString(kv.first, kv.second, &value))
932 ObjectLock olock(query.Fields);
934 for (const Dictionary::Pair& kv : query.Fields) {
937 if (kv.second.IsEmpty() && !kv.second.IsString())
940 if (!FieldToEscapedString(kv.first, kv.second, &value))
948 void IdoMysqlConnection::InternalExecuteMultipleQueries(const std::vector<DbQuery>& queries)
958 for (const DbQuery& query : queries) {
959 ASSERT(query.Type == DbQueryNewTransaction || query.Category != DbCatInvalid);
961 if (!CanExecuteQuery(query)) {
963 #ifdef I2_DEBUG /* I2_DEBUG */
964 Log(LogDebug, "IdoMysqlConnection")
965 << "Scheduling multiple execute query task again: Cannot execute query now. Type '"
966 << query.Type << "', table '" << query.Table << "', queue size: '" << GetPendingQueryCount() << "'.";
967 #endif /* I2_DEBUG */
969 m_QueryQueue.Enqueue(std::bind(&IdoMysqlConnection::InternalExecuteMultipleQueries, this, queries), query.Priority);
974 for (const DbQuery& query : queries) {
975 InternalExecuteQuery(query);
979 void IdoMysqlConnection::InternalExecuteQuery(const DbQuery& query, int typeOverride)
989 if (query.Type == DbQueryNewTransaction) {
990 InternalNewTransaction();
994 /* check whether we're allowed to execute the query first */
995 if (GetCategoryFilter() != DbCatEverything && (query.Category & GetCategoryFilter()) == 0)
998 if (query.Object && query.Object->GetObject()->GetExtension("agent_check").ToBool())
1001 /* check if there are missing object/insert ids and re-enqueue the query */
1002 if (!CanExecuteQuery(query)) {
1004 #ifdef I2_DEBUG /* I2_DEBUG */
1005 Log(LogDebug, "IdoMysqlConnection")
1006 << "Scheduling execute query task again: Cannot execute query now. Type '"
1007 << typeOverride << "', table '" << query.Table << "', queue size: '" << GetPendingQueryCount() << "'.";
1008 #endif /* I2_DEBUG */
1010 m_QueryQueue.Enqueue(std::bind(&IdoMysqlConnection::InternalExecuteQuery, this, query, typeOverride), query.Priority);
1014 std::ostringstream qbuf, where;
1017 if (query.WhereCriteria) {
1020 ObjectLock olock(query.WhereCriteria);
1024 for (const Dictionary::Pair& kv : query.WhereCriteria) {
1025 if (!FieldToEscapedString(kv.first, kv.second, &value)) {
1027 #ifdef I2_DEBUG /* I2_DEBUG */
1028 Log(LogDebug, "IdoMysqlConnection")
1029 << "Scheduling execute query task again: Cannot execute query now. Type '"
1030 << typeOverride << "', table '" << query.Table << "', queue size: '" << GetPendingQueryCount() << "'.";
1031 #endif /* I2_DEBUG */
1033 m_QueryQueue.Enqueue(std::bind(&IdoMysqlConnection::InternalExecuteQuery, this, query, -1), query.Priority);
1040 where << kv.first << " = " << value;
1047 type = (typeOverride != -1) ? typeOverride : query.Type;
1049 bool upsert = false;
1051 if ((type & DbQueryInsert) && (type & DbQueryUpdate)) {
1055 if (query.ConfigUpdate)
1056 hasid = GetConfigUpdate(query.Object);
1057 else if (query.StatusUpdate)
1058 hasid = GetStatusUpdate(query.Object);
1064 type = DbQueryUpdate;
1067 if ((type & DbQueryInsert) && (type & DbQueryDelete)) {
1068 std::ostringstream qdel;
1069 qdel << "DELETE FROM " << GetTablePrefix() << query.Table << where.str();
1070 AsyncQuery(qdel.str());
1072 type = DbQueryInsert;
1077 qbuf << "INSERT INTO " << GetTablePrefix() << query.Table;
1080 qbuf << "UPDATE " << GetTablePrefix() << query.Table << " SET";
1083 qbuf << "DELETE FROM " << GetTablePrefix() << query.Table;
1086 VERIFY(!"Invalid query type.");
1089 if (type == DbQueryInsert || type == DbQueryUpdate) {
1090 std::ostringstream colbuf, valbuf;
1092 if (type == DbQueryUpdate && query.Fields->GetLength() == 0)
1095 ObjectLock olock(query.Fields);
1098 for (const Dictionary::Pair& kv : query.Fields) {
1101 if (kv.second.IsEmpty() && !kv.second.IsString())
1104 if (!FieldToEscapedString(kv.first, kv.second, &value)) {
1106 #ifdef I2_DEBUG /* I2_DEBUG */
1107 Log(LogDebug, "IdoMysqlConnection")
1108 << "Scheduling execute query task again: Cannot extract required INSERT/UPDATE fields, key '"
1109 << kv.first << "', val '" << kv.second << "', type " << typeOverride << ", table '" << query.Table << "'.";
1110 #endif /* I2_DEBUG */
1112 m_QueryQueue.Enqueue(std::bind(&IdoMysqlConnection::InternalExecuteQuery, this, query, -1), query.Priority);
1116 if (type == DbQueryInsert) {
1128 qbuf << " " << kv.first << " = " << value;
1135 if (type == DbQueryInsert)
1136 qbuf << " (" << colbuf.str() << ") VALUES (" << valbuf.str() << ")";
1139 if (type != DbQueryInsert)
1140 qbuf << where.str();
1142 AsyncQuery(qbuf.str(), std::bind(&IdoMysqlConnection::FinishExecuteQuery, this, query, type, upsert));
1145 void IdoMysqlConnection::FinishExecuteQuery(const DbQuery& query, int type, bool upsert)
1147 if (upsert && GetAffectedRows() == 0) {
1149 #ifdef I2_DEBUG /* I2_DEBUG */
1150 Log(LogDebug, "IdoMysqlConnection")
1151 << "Rescheduling DELETE/INSERT query: Upsert UPDATE did not affect rows, type " << type << ", table '" << query.Table << "'.";
1152 #endif /* I2_DEBUG */
1154 m_QueryQueue.Enqueue(std::bind(&IdoMysqlConnection::InternalExecuteQuery, this, query, DbQueryDelete | DbQueryInsert), query.Priority);
1159 if (type == DbQueryInsert && query.Object) {
1160 if (query.ConfigUpdate) {
1161 SetInsertID(query.Object, GetLastInsertID());
1162 SetConfigUpdate(query.Object, true);
1163 } else if (query.StatusUpdate)
1164 SetStatusUpdate(query.Object, true);
1167 if (type == DbQueryInsert && query.Table == "notifications" && query.NotificationInsertID)
1168 query.NotificationInsertID->SetValue(static_cast<long>(GetLastInsertID()));
1171 void IdoMysqlConnection::CleanUpExecuteQuery(const String& table, const String& time_column, double max_age)
1176 #ifdef I2_DEBUG /* I2_DEBUG */
1177 Log(LogDebug, "IdoMysqlConnection")
1178 << "Rescheduling cleanup query for table '" << table << "' and column '"
1179 << time_column << "'. max_age is set to '" << max_age << "'.";
1180 #endif /* I2_DEBUG */
1182 m_QueryQueue.Enqueue(std::bind(&IdoMysqlConnection::InternalCleanUpExecuteQuery, this, table, time_column, max_age), PriorityLow, true);
1185 void IdoMysqlConnection::InternalCleanUpExecuteQuery(const String& table, const String& time_column, double max_age)
1187 AssertOnWorkQueue();
1192 if (!GetConnected())
1195 AsyncQuery("DELETE FROM " + GetTablePrefix() + table + " WHERE instance_id = " +
1196 Convert::ToString(static_cast<long>(m_InstanceID)) + " AND " + time_column +
1197 " < FROM_UNIXTIME(" + Convert::ToString(static_cast<long>(max_age)) + ")");
1200 void IdoMysqlConnection::FillIDCache(const DbType::Ptr& type)
1202 String query = "SELECT " + type->GetIDColumn() + " AS object_id, " + type->GetTable() + "_id, config_hash FROM " + GetTablePrefix() + type->GetTable() + "s";
1203 IdoMysqlResult result = Query(query);
1205 Dictionary::Ptr row;
1207 while ((row = FetchRow(result))) {
1208 DbReference dbref(row->Get("object_id"));
1209 SetInsertID(type, dbref, DbReference(row->Get(type->GetTable() + "_id")));
1210 SetConfigHash(type, dbref, row->Get("config_hash"));
1214 int IdoMysqlConnection::GetPendingQueryCount() const
1216 return m_QueryQueue.GetLength();