tyler.durden/icinga2
1
0
Fork 0
mirror of https://github.com/Icinga/icinga2.git synced 2025-07-25 22:54:57 +02:00
Code Issues Packages Projects Releases Wiki Activity

Implement support for the new config/state schema

Browse Source 
fixes #5671
This commit is contained in:
Gunnar Beutner 2017-10-12 08:51:13 +02:00 committed by Michael Friedrich
parent 1725038ca8
commit 06211c3ac7
3 changed files with 214 additions and 149 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

335
lib/redis/rediswriter-status.cpp
View File

@ -26,6 +26,7 @@
#include "base/serializer.hpp" #include "base/serializer.hpp"
#include "base/tlsutility.hpp" #include "base/tlsutility.hpp"
#include "base/initialize.hpp" #include "base/initialize.hpp"
#include "base/convert.hpp"
using namespace icinga; using namespace icinga;
@ -49,7 +50,6 @@ void RedisWriter::ConfigStaticInitialize(void)
{ {
/* triggered in ProcessCheckResult(), requires UpdateNextCheck() to be called before */ /* triggered in ProcessCheckResult(), requires UpdateNextCheck() to be called before */
ConfigObject::OnStateChanged.connect(boost::bind(&RedisWriter::StateChangedHandler, _1)); ConfigObject::OnStateChanged.connect(boost::bind(&RedisWriter::StateChangedHandler, _1));
CustomVarObject::OnVarsChanged.connect(boost::bind(&RedisWriter::VarsChangedHandler, _1));
/* triggered on create, update and delete objects */ /* triggered on create, update and delete objects */
ConfigObject::OnActiveChanged.connect(boost::bind(&RedisWriter::VersionChangedHandler, _1)); ConfigObject::OnActiveChanged.connect(boost::bind(&RedisWriter::VersionChangedHandler, _1));
@ -62,8 +62,57 @@ void RedisWriter::UpdateAllConfigObjects(void)
double startTime = Utility::GetTime(); double startTime = Utility::GetTime();
//TODO: "Publish" the config dump by adding another event, globally or by object std::vector<String> deleteQuery({ "DEL" });
ExecuteQuery({ "MULTI" }); long long cursor = 0;
const String keyPrefix = "icinga:config:";
do {
boost::shared_ptr<redisReply> reply = ExecuteQuery({ "SCAN", Convert::ToString(cursor), "MATCH", keyPrefix + "*", "COUNT", "1000" });
VERIFY(reply->type == REDIS_REPLY_ARRAY);
VERIFY(reply->elements % 2 == 0);
redisReply *cursorReply = reply->element[0];
cursor = Convert::ToLong(cursorReply->str);
redisReply *keysReply = reply->element[1];
for (size_t i = 0; i < keysReply->elements; i++) {
redisReply *keyReply = keysReply->element[i];
VERIFY(keyReply->type == REDIS_REPLY_STRING);
String key = keyReply->str;
String namePair = key.SubStr(keyPrefix.GetLength());
String::SizeType pos = namePair.FindFirstOf(":");
if (pos == String::NPos)
continue;
String type = namePair.SubStr(0, pos);
String name = namePair.SubStr(pos + 1);
Type::Ptr ptype = Type::GetByName(type);
if (!ptype)
continue;
ConfigType *ctype = dynamic_cast<ConfigType *>(ptype.get());
if (!ctype)
continue;
if (ctype->GetObject(name))
continue;
deleteQuery.push_back("icinga:config:" + type + ":" + name);
deleteQuery.push_back("icinga:status:" + type + ":" + name);
}
} while (cursor != 0);
if (deleteQuery.size() > 1)
ExecuteQuery(deleteQuery);
for (const Type::Ptr& type : Type::GetAllTypes()) { for (const Type::Ptr& type : Type::GetAllTypes()) {
ConfigType *ctype = dynamic_cast<ConfigType *>(type.get()); ConfigType *ctype = dynamic_cast<ConfigType *>(type.get());
@ -77,21 +126,19 @@ void RedisWriter::UpdateAllConfigObjects(void)
/* fetch all objects and dump them */ /* fetch all objects and dump them */
for (const ConfigObject::Ptr& object : ctype->GetObjects()) { for (const ConfigObject::Ptr& object : ctype->GetObjects()) {
SendConfigUpdate(object, typeName); SendConfigUpdate(object, false);
SendStatusUpdate(object, typeName); SendStatusUpdate(object, false);
} }
/* publish config type dump finished */ /* publish config type dump finished */
ExecuteQuery({ "PUBLISH", "icinga:config:dump", typeName }); ExecuteQuery({ "PUBLISH", "icinga:config:dump", typeName });
} }
ExecuteQuery({ "EXEC" });
Log(LogInformation, "RedisWriter") Log(LogInformation, "RedisWriter")
<< "Initial config/status dump finished in " << Utility::GetTime() - startTime << " seconds."; << "Initial config/status dump finished in " << Utility::GetTime() - startTime << " seconds.";
} }
void RedisWriter::SendConfigUpdate(const ConfigObject::Ptr& object, const String& typeName, bool runtimeUpdate) void RedisWriter::SendConfigUpdate(const ConfigObject::Ptr& object, bool useTransaction, bool runtimeUpdate)
{ {
AssertOnWorkQueue(); AssertOnWorkQueue();
@ -104,55 +151,62 @@ void RedisWriter::SendConfigUpdate(const ConfigObject::Ptr& object, const String
return; return;
*/ */
/* Serialize config object attributes */ if (useTransaction)
Dictionary::Ptr objectAttrs = SerializeObjectAttrs(object, FAConfig); ExecuteQuery({ "MULTI" });
String jsonBody = JsonEncode(objectAttrs); UpdateObjectAttrs("icinga:config:", object, FAConfig);
String objectName = object->GetName(); // /* Serialize config object attributes */
// Dictionary::Ptr objectAttrs = SerializeObjectAttrs(object, FAConfig);
//
// String jsonBody = JsonEncode(objectAttrs);
//
// String objectName = object->GetName();
//
// ExecuteQuery({ "HSET", "icinga:config:" + typeName, objectName, jsonBody });
//
// /* check sums */
// /* hset icinga:config:Host:checksums localhost { "name_checksum": "...", "properties_checksum": "...", "groups_checksum": "...", "vars_checksum": null } */
// Dictionary::Ptr checkSum = new Dictionary();
//
// checkSum->Set("name_checksum", CalculateCheckSumString(object->GetName()));
//
// // TODO: move this elsewhere
// Checkable::Ptr checkable = dynamic_pointer_cast<Checkable>(object);
//
// if (checkable) {
// Host::Ptr host;
// Service::Ptr service;
//
// tie(host, service) = GetHostService(checkable);
//
// if (service)
// checkSum->Set("groups_checksum", CalculateCheckSumGroups(service->GetGroups()));
// else
// checkSum->Set("groups_checksum", CalculateCheckSumGroups(host->GetGroups()));
// }
//
// checkSum->Set("properties_checksum", CalculateCheckSumProperties(object));
//
// CustomVarObject::Ptr customVarObject = dynamic_pointer_cast<CustomVarObject>(object);
//
// if (customVarObject)
// checkSum->Set("vars_checksum", CalculateCheckSumVars(customVarObject));
//
// String checkSumBody = JsonEncode(checkSum);
//
// ExecuteQuery({ "HSET", "icinga:config:" + typeName + ":checksum", objectName, checkSumBody });
ExecuteQuery({ "HSET", "icinga:config:" + typeName, objectName, jsonBody }); if (runtimeUpdate) {
Type::Ptr type = object->GetReflectionType();
/* check sums */ ExecuteQuery({ "PUBLISH", "icinga:config:update", type->GetName() + ":" + object->GetName() });
/* hset icinga:config:Host:checksums localhost { "name_checksum": "...", "properties_checksum": "...", "groups_checksum": "...", "vars_checksum": null } */
Dictionary::Ptr checkSum = new Dictionary();
checkSum->Set("name_checksum", CalculateCheckSumString(object->GetName()));
// TODO: move this elsewhere
Checkable::Ptr checkable = dynamic_pointer_cast<Checkable>(object);
if (checkable) {
Host::Ptr host;
Service::Ptr service;
tie(host, service) = GetHostService(checkable);
if (service)
checkSum->Set("groups_checksum", CalculateCheckSumGroups(service->GetGroups()));
else
checkSum->Set("groups_checksum", CalculateCheckSumGroups(host->GetGroups()));
} }
checkSum->Set("properties_checksum", CalculateCheckSumProperties(object)); if (useTransaction)
ExecuteQuery({ "EXEC" });
CustomVarObject::Ptr customVarObject = dynamic_pointer_cast<CustomVarObject>(object);
if (customVarObject)
checkSum->Set("vars_checksum", CalculateCheckSumVars(customVarObject));
String checkSumBody = JsonEncode(checkSum);
ExecuteQuery({ "HSET", "icinga:config:" + typeName + ":checksum", objectName, checkSumBody });
/* publish runtime updated objects immediately */
if (!runtimeUpdate)
return;
ExecuteQuery({ "PUBLISH", "icinga:config:update", typeName + ":" + objectName + "!" + checkSumBody });
} }
void RedisWriter::SendConfigDelete(const ConfigObject::Ptr& object, const String& typeName) void RedisWriter::SendConfigDelete(const ConfigObject::Ptr& object)
{ {
AssertOnWorkQueue(); AssertOnWorkQueue();
@ -160,16 +214,18 @@ void RedisWriter::SendConfigDelete(const ConfigObject::Ptr& object, const String
if (!m_Context) if (!m_Context)
return; return;
String typeName = object->GetReflectionType()->GetName();
String objectName = object->GetName(); String objectName = object->GetName();
ExecuteQuery({ "HDEL", "icinga:config:" + typeName, objectName }); ExecuteQueries({
ExecuteQuery({ "HDEL", "icinga:config:" + typeName + ":checksum", objectName }); { "DEL", "icinga:config:" + typeName, objectName },
ExecuteQuery({ "HDEL", "icinga:status:" + typeName, objectName }); { "DEL", "icinga:status:" + typeName, objectName },
{ "PUBLISH", "icinga:config:delete", typeName + ":" + objectName }
});
ExecuteQuery({ "PUBLISH", "icinga:config:delete", typeName + ":" + objectName });
} }
void RedisWriter::SendStatusUpdate(const ConfigObject::Ptr& object, const String& typeName) void RedisWriter::SendStatusUpdate(const ConfigObject::Ptr& object, bool useTransaction)
{ {
AssertOnWorkQueue(); AssertOnWorkQueue();
@ -177,82 +233,90 @@ void RedisWriter::SendStatusUpdate(const ConfigObject::Ptr& object, const String
if (!m_Context) if (!m_Context)
return; return;
/* Serialize config object attributes */ if (useTransaction)
Dictionary::Ptr objectAttrs = SerializeObjectAttrs(object, FAState); ExecuteQuery({ "MULTI" });
String jsonBody = JsonEncode(objectAttrs); UpdateObjectAttrs("icinga:status:", object, FAState);
String objectName = object->GetName(); if (useTransaction)
ExecuteQuery({ "EXEC" });
ExecuteQuery({ "HSET", "icinga:status:" + typeName, objectName, jsonBody }); // /* Serialize config object attributes */
// Dictionary::Ptr objectAttrs = SerializeObjectAttrs(object, FAState);
/* Icinga DB part for Icinga Web 2 */ //
Checkable::Ptr checkable = dynamic_pointer_cast<Checkable>(object); // String jsonBody = JsonEncode(objectAttrs);
//
if (checkable) { // String objectName = object->GetName();
Dictionary::Ptr attrs = new Dictionary(); //
String tableName; // ExecuteQuery({ "HSET", "icinga:status:" + typeName, objectName, jsonBody });
String objectCheckSum = CalculateCheckSumString(objectName); //
// /* Icinga DB part for Icinga Web 2 */
Host::Ptr host; // Checkable::Ptr checkable = dynamic_pointer_cast<Checkable>(object);
Service::Ptr service; //
// if (checkable) {
tie(host, service) = GetHostService(checkable); // Dictionary::Ptr attrs = new Dictionary();
// String tableName;
if (service) { // String objectCheckSum = CalculateCheckSumString(objectName);
tableName = "servicestate"; //
attrs->Set("service_checksum", objectCheckSum); // Host::Ptr host;
attrs->Set("host_checksum", CalculateCheckSumString(host->GetName())); // Service::Ptr service;
} else { //
tableName = "hoststate"; // tie(host, service) = GetHostService(checkable);
attrs->Set("host_checksum", objectCheckSum); //
} // if (service) {
// tableName = "servicestate";
attrs->Set("last_check", checkable->GetLastCheck()); // attrs->Set("service_checksum", objectCheckSum);
attrs->Set("next_check", checkable->GetNextCheck()); // attrs->Set("host_checksum", CalculateCheckSumString(host->GetName()));
// } else {
attrs->Set("severity", checkable->GetSeverity()); // tableName = "hoststate";
// attrs->Set("host_checksum", objectCheckSum);
/* // }
'host_checksum' => null, //
'command' => null, // JSON, array // attrs->Set("last_check", checkable->GetLastCheck());
'execution_start' => null, // attrs->Set("next_check", checkable->GetNextCheck());
'execution_end' => null, //
'schedule_start' => null, // attrs->Set("severity", checkable->GetSeverity());
'schedule_end' => null, //
'exit_status' => null, ///*
'output' => null, // 'host_checksum' => null,
'performance_data' => null, // JSON, array // 'command' => null, // JSON, array
// 'execution_start' => null,
// 'execution_end' => null,
10.0.3.12:6379> keys icinga:hoststate.* // 'schedule_start' => null,
1) "icinga:hoststate.~\xf5a\x91+\x03\x97\x99\xb5(\x16 CYm\xb1\xdf\x85\xa2\xcb" // 'schedule_end' => null,
10.0.3.12:6379> get "icinga:hoststate.~\xf5a\x91+\x03\x97\x99\xb5(\x16 CYm\xb1\xdf\x85\xa2\xcb" // 'exit_status' => null,
"{\"command\":[\"\\/usr\\/lib\\/nagios\\/plugins\\/check_ping\",\"-H\",\"127.0.0.1\",\"-c\",\"5000,100%\",\"-w\",\"3000,80%\"],\"execution_start\":1492007581.7624,\"execution_end\":1492007585.7654,\"schedule_start\":1492007581.7609,\"schedule_end\":1492007585.7655,\"exit_status\":0,\"output\":\"PING OK - Packet loss = 0%, RTA = 0.08 ms\",\"performance_data\":[\"rta=0.076000ms;3000.000000;5000.000000;0.000000\",\"pl=0%;80;100;0\"]}" // 'output' => null,
// 'performance_data' => null, // JSON, array
*/ //
//
CheckResult::Ptr cr = checkable->GetLastCheckResult(); //10.0.3.12:6379> keys icinga:hoststate.*
//1) "icinga:hoststate.~\xf5a\x91+\x03\x97\x99\xb5(\x16 CYm\xb1\xdf\x85\xa2\xcb"
if (cr) { //10.0.3.12:6379> get "icinga:hoststate.~\xf5a\x91+\x03\x97\x99\xb5(\x16 CYm\xb1\xdf\x85\xa2\xcb"
attrs->Set("command", JsonEncode(cr->GetCommand())); //"{\"command\":[\"\\/usr\\/lib\\/nagios\\/plugins\\/check_ping\",\"-H\",\"127.0.0.1\",\"-c\",\"5000,100%\",\"-w\",\"3000,80%\"],\"execution_start\":1492007581.7624,\"execution_end\":1492007585.7654,\"schedule_start\":1492007581.7609,\"schedule_end\":1492007585.7655,\"exit_status\":0,\"output\":\"PING OK - Packet loss = 0%, RTA = 0.08 ms\",\"performance_data\":[\"rta=0.076000ms;3000.000000;5000.000000;0.000000\",\"pl=0%;80;100;0\"]}"
attrs->Set("execution_start", cr->GetExecutionStart()); //
attrs->Set("execution_end", cr->GetExecutionEnd()); //*/
attrs->Set("schedule_start", cr->GetScheduleStart()); //
attrs->Set("schedule_end", cr->GetScheduleStart()); // CheckResult::Ptr cr = checkable->GetLastCheckResult();
attrs->Set("exit_status", cr->GetExitStatus()); //
attrs->Set("output", cr->GetOutput()); // if (cr) {
attrs->Set("performance_data", JsonEncode(cr->GetPerformanceData())); // attrs->Set("command", JsonEncode(cr->GetCommand()));
} // attrs->Set("execution_start", cr->GetExecutionStart());
// attrs->Set("execution_end", cr->GetExecutionEnd());
String jsonAttrs = JsonEncode(attrs); // attrs->Set("schedule_start", cr->GetScheduleStart());
String key = "icinga:" + tableName + "." + objectCheckSum; // attrs->Set("schedule_end", cr->GetScheduleStart());
ExecuteQuery({ "SET", key, jsonAttrs }); // attrs->Set("exit_status", cr->GetExitStatus());
// attrs->Set("output", cr->GetOutput());
/* expire in check_interval * attempts + timeout + some more seconds */ // attrs->Set("performance_data", JsonEncode(cr->GetPerformanceData()));
double expireTime = checkable->GetCheckInterval() * checkable->GetMaxCheckAttempts() + 60; // }
ExecuteQuery({ "EXPIRE", key, String(expireTime) }); //
} // String jsonAttrs = JsonEncode(attrs);
// String key = "icinga:" + tableName + "." + objectCheckSum;
// ExecuteQuery({ "SET", key, jsonAttrs });
//
// /* expire in check_interval * attempts + timeout + some more seconds */
// double expireTime = checkable->GetCheckInterval() * checkable->GetMaxCheckAttempts() + 60;
// ExecuteQuery({ "EXPIRE", key, String(expireTime) });
// }
} }
void RedisWriter::StateChangedHandler(const ConfigObject::Ptr& object) void RedisWriter::StateChangedHandler(const ConfigObject::Ptr& object)
@ -260,16 +324,7 @@ void RedisWriter::StateChangedHandler(const ConfigObject::Ptr& object)
Type::Ptr type = object->GetReflectionType(); Type::Ptr type = object->GetReflectionType();
for (const RedisWriter::Ptr& rw : ConfigType::GetObjectsByType<RedisWriter>()) { for (const RedisWriter::Ptr& rw : ConfigType::GetObjectsByType<RedisWriter>()) {
rw->m_WorkQueue.Enqueue(boost::bind(&RedisWriter::SendStatusUpdate, rw, object, type->GetName())); rw->m_WorkQueue.Enqueue(boost::bind(&RedisWriter::SendStatusUpdate, rw, object, true));
}
}
void RedisWriter::VarsChangedHandler(const ConfigObject::Ptr& object)
{
Type::Ptr type = object->GetReflectionType();
for (const RedisWriter::Ptr& rw : ConfigType::GetObjectsByType<RedisWriter>()) {
rw->m_WorkQueue.Enqueue(boost::bind(&RedisWriter::SendConfigUpdate, rw, object, type->GetName(), true));
} }
} }
@ -280,12 +335,12 @@ void RedisWriter::VersionChangedHandler(const ConfigObject::Ptr& object)
if (object->IsActive()) { if (object->IsActive()) {
/* Create or update the object config */ /* Create or update the object config */
for (const RedisWriter::Ptr& rw : ConfigType::GetObjectsByType<RedisWriter>()) { for (const RedisWriter::Ptr& rw : ConfigType::GetObjectsByType<RedisWriter>()) {
rw->m_WorkQueue.Enqueue(boost::bind(&RedisWriter::SendConfigUpdate, rw.get(), object, type->GetName(), true)); rw->m_WorkQueue.Enqueue(boost::bind(&RedisWriter::SendConfigUpdate, rw.get(), object, true, true));
} }
} else if (!object->IsActive() && object->GetExtension("ConfigObjectDeleted")) { /* same as in apilistener-configsync.cpp */ } else if (!object->IsActive() && object->GetExtension("ConfigObjectDeleted")) { /* same as in apilistener-configsync.cpp */
/* Delete object config */ /* Delete object config */
for (const RedisWriter::Ptr& rw : ConfigType::GetObjectsByType<RedisWriter>()) { for (const RedisWriter::Ptr& rw : ConfigType::GetObjectsByType<RedisWriter>()) {
rw->m_WorkQueue.Enqueue(boost::bind(&RedisWriter::SendConfigDelete, rw.get(), object, type->GetName())); rw->m_WorkQueue.Enqueue(boost::bind(&RedisWriter::SendConfigDelete, rw.get(), object));
} }
} }
} }

19
lib/redis/rediswriter-utility.cpp
View File

@ -86,11 +86,18 @@ String RedisWriter::HashValue(const Value& value)
return SHA1(JsonEncode(temp)); return SHA1(JsonEncode(temp));
} }
Dictionary::Ptr RedisWriter::SerializeObjectAttrs(const Object::Ptr& object, int fieldType) void RedisWriter::UpdateObjectAttrs(const String& keyPrefix, const ConfigObject::Ptr& object, int fieldType)
{ {
Type::Ptr type = object->GetReflectionType(); Type::Ptr type = object->GetReflectionType();
Dictionary::Ptr resultAttrs = new Dictionary(); String typeName = type->GetName();
String objectName = object->GetName();
std::vector<std::vector<String> > queries;
queries.push_back({ "DEL", keyPrefix + object->GetName() });
std::vector<String> hmsetCommand({ "HMSET", keyPrefix + typeName + ":" + objectName });
for (int fid = 0; fid < type->GetFieldCount(); fid++) { for (int fid = 0; fid < type->GetFieldCount(); fid++) {
Field field = type->GetFieldInfo(fid); Field field = type->GetFieldInfo(fid);
@ -108,10 +115,14 @@ Dictionary::Ptr RedisWriter::SerializeObjectAttrs(const Object::Ptr& object, int
if (field.Attributes & FANavigation && !(field.Attributes & (FAConfig | FAState))) if (field.Attributes & FANavigation && !(field.Attributes & (FAConfig | FAState)))
continue; continue;
hmsetCommand.push_back(field.Name);
Value sval = Serialize(val); Value sval = Serialize(val);
resultAttrs->Set(field.Name, sval); hmsetCommand.push_back(JsonEncode(sval));
} }
return resultAttrs; queries.push_back(hmsetCommand);
ExecuteQueries(queries);
} }

9
lib/redis/rediswriter.hpp
View File

@ -64,9 +64,10 @@ private:
/* config & status dump */ /* config & status dump */
void UpdateAllConfigObjects(void); void UpdateAllConfigObjects(void);
void SendConfigUpdate(const ConfigObject::Ptr& object, const String& typeName, bool runtimeUpdate = false); void SendConfigUpdate(const ConfigObject::Ptr& object, bool useTransaction, bool runtimeUpdate = false);
void SendConfigDelete(const ConfigObject::Ptr& object, const String& typeName); void SendConfigDelete(const ConfigObject::Ptr& object);
void SendStatusUpdate(const ConfigObject::Ptr& object, const String& typeName); void SendStatusUpdate(const ConfigObject::Ptr& object, bool useTransaction);
void UpdateObjectAttrs(const String& keyPrefix, const ConfigObject::Ptr& object, int fieldType);
/* utilities */ /* utilities */
static String FormatCheckSumBinary(const String& str); static String FormatCheckSumBinary(const String& str);
@ -77,10 +78,8 @@ private:
static String CalculateCheckSumVars(const CustomVarObject::Ptr& object); static String CalculateCheckSumVars(const CustomVarObject::Ptr& object);
static String HashValue(const Value& value); static String HashValue(const Value& value);
static Dictionary::Ptr SerializeObjectAttrs(const Object::Ptr& object, int fieldType);
static void StateChangedHandler(const ConfigObject::Ptr& object); static void StateChangedHandler(const ConfigObject::Ptr& object);
static void VarsChangedHandler(const ConfigObject::Ptr& object);
static void VersionChangedHandler(const ConfigObject::Ptr& object); static void VersionChangedHandler(const ConfigObject::Ptr& object);
void AssertOnWorkQueue(void); void AssertOnWorkQueue(void);