From 96fa29793c15692eddc462c61078b37ce17879e2 Mon Sep 17 00:00:00 2001 From: Michael Friedrich Date: Tue, 15 Dec 2015 13:44:58 +0100 Subject: [PATCH] Avoid duplicate config and status updates on startup fixes #10765 --- lib/db_ido/dbconnection.cpp | 6 +- lib/db_ido_mysql/idomysqlconnection.cpp | 21 +- lib/db_ido_pgsql/idopgsqlconnection.cpp | 375 ++++++++++++------------ 3 files changed, 203 insertions(+), 199 deletions(-) diff --git a/lib/db_ido/dbconnection.cpp b/lib/db_ido/dbconnection.cpp index 16b45db07..d86e87e7e 100644 --- a/lib/db_ido/dbconnection.cpp +++ b/lib/db_ido/dbconnection.cpp @@ -364,14 +364,14 @@ void DbConnection::UpdateObject(const ConfigObject::Ptr& object) DbObject::Ptr dbobj = DbObject::GetOrCreateByObject(object); if (dbobj) { + bool dbActive = GetObjectActive(dbobj); bool active = object->IsActive(); - if (active) { + if (active && !dbActive) { ActivateObject(dbobj); - dbobj->SendConfigUpdate(); dbobj->SendStatusUpdate(); - } else + } else if (!active && dbActive) DeactivateObject(dbobj); } } diff --git a/lib/db_ido_mysql/idomysqlconnection.cpp b/lib/db_ido_mysql/idomysqlconnection.cpp index b1e1fc196..18b55b4ce 100644 --- a/lib/db_ido_mysql/idomysqlconnection.cpp +++ b/lib/db_ido_mysql/idomysqlconnection.cpp @@ -171,8 +171,6 @@ void IdoMysqlConnection::Reconnect(void) SetShouldConnect(true); - std::vector active_dbobjs; - bool reconnect = false; if (GetConnected()) { @@ -351,6 +349,8 @@ void IdoMysqlConnection::Reconnect(void) q1buf << "SELECT object_id, objecttype_id, name1, name2, is_active FROM " + GetTablePrefix() + "objects WHERE instance_id = " << static_cast(m_InstanceID); result = Query(q1buf.str()); + std::vector activeDbObjs; + while ((row = FetchRow(result))) { DbType::Ptr dbtype = DbType::GetByID(row->Get("objecttype_id")); @@ -359,26 +359,29 @@ void IdoMysqlConnection::Reconnect(void) DbObject::Ptr dbobj = dbtype->GetOrCreateObjectByName(row->Get("name1"), row->Get("name2")); SetObjectID(dbobj, DbReference(row->Get("object_id"))); - SetObjectActive(dbobj, row->Get("is_active")); + bool active = row->Get("is_active"); + SetObjectActive(dbobj, active); - if (GetObjectActive(dbobj)) - active_dbobjs.push_back(dbobj); + if (active) + activeDbObjs.push_back(dbobj); } Query("BEGIN"); - UpdateAllObjects(); - - /* deactivate all deleted configuration objects */ - BOOST_FOREACH(const DbObject::Ptr& dbobj, active_dbobjs) { + BOOST_FOREACH(const DbObject::Ptr& dbobj, activeDbObjs) { if (dbobj->GetObject() == NULL) { Log(LogNotice, "IdoMysqlConnection") << "Deactivate deleted object name1: '" << dbobj->GetName1() << "' name2: '" << dbobj->GetName2() + "'."; DeactivateObject(dbobj); + } else { + dbobj->SendConfigUpdate(); + dbobj->SendStatusUpdate(); } } + UpdateAllObjects(); + /* delete all customvariables without current session token */ ClearCustomVarTable("customvariables"); ClearCustomVarTable("customvariablestatus"); diff --git a/lib/db_ido_pgsql/idopgsqlconnection.cpp b/lib/db_ido_pgsql/idopgsqlconnection.cpp index 29a058848..f1adafb04 100644 --- a/lib/db_ido_pgsql/idopgsqlconnection.cpp +++ b/lib/db_ido_pgsql/idopgsqlconnection.cpp @@ -168,209 +168,210 @@ void IdoPgsqlConnection::Reconnect(void) SetShouldConnect(true); - std::vector active_dbobjs; + bool reconnect = false; - { - bool reconnect = false; - - if (GetConnected()) { - /* Check if we're really still connected */ - try { - Query("SELECT 1"); - return; - } catch (const std::exception&) { - PQfinish(m_Connection); - SetConnected(false); - reconnect = true; - } - } - - ClearIDCache(); - - String ihost, iport, iuser, ipasswd, idb; - const char *host, *port, *user , *passwd, *db; - - ihost = GetHost(); - iport = GetPort(); - iuser = GetUser(); - ipasswd = GetPassword(); - idb = GetDatabase(); - - host = (!ihost.IsEmpty()) ? ihost.CStr() : NULL; - port = (!iport.IsEmpty()) ? iport.CStr() : NULL; - user = (!iuser.IsEmpty()) ? iuser.CStr() : NULL; - passwd = (!ipasswd.IsEmpty()) ? ipasswd.CStr() : NULL; - db = (!idb.IsEmpty()) ? idb.CStr() : NULL; - - m_Connection = PQsetdbLogin(host, port, NULL, NULL, db, user, passwd); - - if (!m_Connection) + if (GetConnected()) { + /* Check if we're really still connected */ + try { + Query("SELECT 1"); return; - - if (PQstatus(m_Connection) != CONNECTION_OK) { - String message = PQerrorMessage(m_Connection); + } catch (const std::exception&) { PQfinish(m_Connection); SetConnected(false); - - Log(LogCritical, "IdoPgsqlConnection") - << "Connection to database '" << db << "' with user '" << user << "' on '" << host << ":" << port - << "' failed: \"" << message << "\""; - - BOOST_THROW_EXCEPTION(std::runtime_error(message)); + reconnect = true; } - - SetConnected(true); - - IdoPgsqlResult result; - - /* explicitely require legacy mode for string escaping in PostgreSQL >= 9.1 - * changing standard_conforming_strings to on by default - */ - if (PQserverVersion(m_Connection) >= 90100) - result = Query("SET standard_conforming_strings TO off"); - - String dbVersionName = "idoutils"; - result = Query("SELECT version FROM " + GetTablePrefix() + "dbversion WHERE name=E'" + Escape(dbVersionName) + "'"); - - Dictionary::Ptr row = FetchRow(result, 0); - - if (!row) { - PQfinish(m_Connection); - SetConnected(false); - - Log(LogCritical, "IdoPgsqlConnection", "Schema does not provide any valid version! Verify your schema installation."); - - Application::Exit(EXIT_FAILURE); - } - - String version = row->Get("version"); - - SetSchemaVersion(version); - - if (Utility::CompareVersion(IDO_COMPAT_SCHEMA_VERSION, version) < 0) { - PQfinish(m_Connection); - SetConnected(false); - - Log(LogCritical, "IdoPgsqlConnection") - << "Schema version '" << version << "' does not match the required version '" - << IDO_COMPAT_SCHEMA_VERSION << "' (or newer)! Please check the upgrade documentation."; - - Application::Exit(EXIT_FAILURE); - } - - String instanceName = GetInstanceName(); - - result = Query("SELECT instance_id FROM " + GetTablePrefix() + "instances WHERE instance_name = E'" + Escape(instanceName) + "'"); - row = FetchRow(result, 0); - - if (!row) { - Query("INSERT INTO " + GetTablePrefix() + "instances (instance_name, instance_description) VALUES (E'" + Escape(instanceName) + "', E'" + Escape(GetInstanceDescription()) + "')"); - m_InstanceID = GetSequenceValue(GetTablePrefix() + "instances", "instance_id"); - } else { - m_InstanceID = DbReference(row->Get("instance_id")); - } - - 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, 0); - - String endpoint_name; - - if (row) - endpoint_name = row->Get("endpoint_name"); - else - Log(LogNotice, "IdoPgsqlConnection", "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 status_update_age = Utility::GetTime() - status_update_time; - - Log(LogNotice, "IdoPgsqlConnection") - << "Last update by '" << endpoint_name << "' was " << status_update_age << "s ago."; - - if (status_update_age < GetFailoverTimeout()) { - PQfinish(m_Connection); - SetConnected(false); - SetShouldConnect(false); - - return; - } - - /* activate the IDO only, if we're authoritative in this zone */ - if (IsPaused()) { - Log(LogNotice, "IdoPgsqlConnection") - << "Local endpoint '" << my_endpoint->GetName() << "' is not authoritative, bailing out."; - - PQfinish(m_Connection); - SetConnected(false); - - return; - } - } - - Log(LogNotice, "IdoPgsqlConnection", "Enabling IDO connection."); - } - - Log(LogInformation, "IdoPgsqlConnection") - << "pgSQL IDO instance id: " << static_cast(m_InstanceID) << " (schema version: '" + version + "')"; - - /* 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(m_InstanceID)) + ", NOW(), NOW(), E'icinga2 db_ido_pgsql', E'" + Escape(Application::GetAppVersion()) - + "', E'" + (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(m_InstanceID); - result = Query(q1buf.str()); - - int index = 0; - while ((row = FetchRow(result, index))) { - index++; - - 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"))); - SetObjectActive(dbobj, row->Get("is_active")); - - if (GetObjectActive(dbobj)) - active_dbobjs.push_back(dbobj); - } - - Query("BEGIN"); } - UpdateAllObjects(); + ClearIDCache(); - /* deactivate all deleted configuration objects */ - BOOST_FOREACH(const DbObject::Ptr& dbobj, active_dbobjs) { + String ihost, iport, iuser, ipasswd, idb; + const char *host, *port, *user , *passwd, *db; + + ihost = GetHost(); + iport = GetPort(); + iuser = GetUser(); + ipasswd = GetPassword(); + idb = GetDatabase(); + + host = (!ihost.IsEmpty()) ? ihost.CStr() : NULL; + port = (!iport.IsEmpty()) ? iport.CStr() : NULL; + user = (!iuser.IsEmpty()) ? iuser.CStr() : NULL; + passwd = (!ipasswd.IsEmpty()) ? ipasswd.CStr() : NULL; + db = (!idb.IsEmpty()) ? idb.CStr() : NULL; + + m_Connection = PQsetdbLogin(host, port, NULL, NULL, db, user, passwd); + + if (!m_Connection) + return; + + if (PQstatus(m_Connection) != CONNECTION_OK) { + String message = PQerrorMessage(m_Connection); + PQfinish(m_Connection); + SetConnected(false); + + Log(LogCritical, "IdoPgsqlConnection") + << "Connection to database '" << db << "' with user '" << user << "' on '" << host << ":" << port + << "' failed: \"" << message << "\""; + + BOOST_THROW_EXCEPTION(std::runtime_error(message)); + } + + SetConnected(true); + + IdoPgsqlResult result; + + /* explicitely require legacy mode for string escaping in PostgreSQL >= 9.1 + * changing standard_conforming_strings to on by default + */ + if (PQserverVersion(m_Connection) >= 90100) + result = Query("SET standard_conforming_strings TO off"); + + String dbVersionName = "idoutils"; + result = Query("SELECT version FROM " + GetTablePrefix() + "dbversion WHERE name=E'" + Escape(dbVersionName) + "'"); + + Dictionary::Ptr row = FetchRow(result, 0); + + if (!row) { + PQfinish(m_Connection); + SetConnected(false); + + Log(LogCritical, "IdoPgsqlConnection", "Schema does not provide any valid version! Verify your schema installation."); + + Application::Exit(EXIT_FAILURE); + } + + String version = row->Get("version"); + + SetSchemaVersion(version); + + if (Utility::CompareVersion(IDO_COMPAT_SCHEMA_VERSION, version) < 0) { + PQfinish(m_Connection); + SetConnected(false); + + Log(LogCritical, "IdoPgsqlConnection") + << "Schema version '" << version << "' does not match the required version '" + << IDO_COMPAT_SCHEMA_VERSION << "' (or newer)! Please check the upgrade documentation."; + + Application::Exit(EXIT_FAILURE); + } + + String instanceName = GetInstanceName(); + + result = Query("SELECT instance_id FROM " + GetTablePrefix() + "instances WHERE instance_name = E'" + Escape(instanceName) + "'"); + row = FetchRow(result, 0); + + if (!row) { + Query("INSERT INTO " + GetTablePrefix() + "instances (instance_name, instance_description) VALUES (E'" + Escape(instanceName) + "', E'" + Escape(GetInstanceDescription()) + "')"); + m_InstanceID = GetSequenceValue(GetTablePrefix() + "instances", "instance_id"); + } else { + m_InstanceID = DbReference(row->Get("instance_id")); + } + + 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, 0); + + String endpoint_name; + + if (row) + endpoint_name = row->Get("endpoint_name"); + else + Log(LogNotice, "IdoPgsqlConnection", "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 status_update_age = Utility::GetTime() - status_update_time; + + Log(LogNotice, "IdoPgsqlConnection") + << "Last update by '" << endpoint_name << "' was " << status_update_age << "s ago."; + + if (status_update_age < GetFailoverTimeout()) { + PQfinish(m_Connection); + SetConnected(false); + SetShouldConnect(false); + + return; + } + + /* activate the IDO only, if we're authoritative in this zone */ + if (IsPaused()) { + Log(LogNotice, "IdoPgsqlConnection") + << "Local endpoint '" << my_endpoint->GetName() << "' is not authoritative, bailing out."; + + PQfinish(m_Connection); + SetConnected(false); + + return; + } + } + + Log(LogNotice, "IdoPgsqlConnection", "Enabling IDO connection."); + } + + Log(LogInformation, "IdoPgsqlConnection") + << "pgSQL IDO instance id: " << static_cast(m_InstanceID) << " (schema version: '" + version + "')"; + + /* 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(m_InstanceID)) + ", NOW(), NOW(), E'icinga2 db_ido_pgsql', E'" + Escape(Application::GetAppVersion()) + + "', E'" + (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(m_InstanceID); + result = Query(q1buf.str()); + + std::vector activeDbObjs; + + int index = 0; + while ((row = FetchRow(result, index))) { + index++; + + 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.push_back(dbobj); + } + + Query("BEGIN"); + + BOOST_FOREACH(const DbObject::Ptr& dbobj, activeDbObjs) { if (dbobj->GetObject() == NULL) { Log(LogNotice, "IdoPgsqlConnection") << "Deactivate deleted object name1: '" << dbobj->GetName1() << "' name2: '" << dbobj->GetName2() + "'."; DeactivateObject(dbobj); + } else { + dbobj->SendConfigUpdate(); + dbobj->SendStatusUpdate(); } } + UpdateAllObjects(); + /* delete all customvariables without current session token */ ClearCustomVarTable("customvariables"); ClearCustomVarTable("customvariablestatus");