tyler.durden
/
icinga2
mirror of https://github.com/Icinga/icinga2.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Use sets for subscriptions 

refs #5656
This commit is contained in:
Gunnar Beutner 2017-10-12 11:47:49 +02:00 committed by Michael Friedrich
parent 06211c3ac7
commit f631bf8cb5
1 changed files with 33 additions and 32 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

41
lib/redis/rediswriter.cpp
View File

@ -32,7 +32,9 @@ REGISTER_TYPE(RedisWriter);
RedisWriter::RedisWriter(void) RedisWriter::RedisWriter(void)
: m_Context(NULL) : m_Context(NULL)
{ } {
m_WorkQueue.SetName("RedisWriter");
}
/** /**
* Starts the component. * Starts the component.
@ -134,6 +136,8 @@ void RedisWriter::TryToReconnect(void)
if (dbIndex != 0) if (dbIndex != 0)
ExecuteQuery({ "SELECT", Convert::ToString(dbIndex) }); ExecuteQuery({ "SELECT", Convert::ToString(dbIndex) });
UpdateSubscriptions();
/* Config dump */ /* Config dump */
m_ConfigDumpInProgress = true; m_ConfigDumpInProgress = true;
@ -151,16 +155,19 @@ void RedisWriter::UpdateSubscriptions(void)
{ {
AssertOnWorkQueue(); AssertOnWorkQueue();
Log(LogInformation, "RedisWriter", "Updating Redis subscriptions");
m_Subscriptions.clear();
if (!m_Context) if (!m_Context)
return; return;
Log(LogInformation, "RedisWriter", "Updating Redis subscriptions");
Array::Ptr keys = new Array();
long long cursor = 0; long long cursor = 0;
String keyPrefix = "icinga:subscription:";
do { do {
boost::shared_ptr<redisReply> reply = ExecuteQuery({ "SCAN", Convert::ToString(cursor), "MATCH", "icinga:subscription:*", "COUNT", "1000" }); boost::shared_ptr<redisReply> reply = ExecuteQuery({ "SCAN", Convert::ToString(cursor), "MATCH", keyPrefix + "*", "COUNT", "1000" });
VERIFY(reply->type == REDIS_REPLY_ARRAY); VERIFY(reply->type == REDIS_REPLY_ARRAY);
VERIFY(reply->elements % 2 == 0); VERIFY(reply->elements % 2 == 0);
@ -173,36 +180,30 @@ void RedisWriter::UpdateSubscriptions(void)
for (size_t i = 0; i < keysReply->elements; i++) { for (size_t i = 0; i < keysReply->elements; i++) {
redisReply *keyReply = keysReply->element[i]; redisReply *keyReply = keysReply->element[i];
VERIFY(keyReply->type == REDIS_REPLY_STRING); VERIFY(keyReply->type == REDIS_REPLY_STRING);
keys->Add(keyReply->str);
}
} while (cursor != 0);
m_Subscriptions.clear(); String key = keyReply->str;
ObjectLock olock(keys);
for (String key : keys) {
try { try {
boost::shared_ptr<redisReply> redisReply = ExecuteQuery({ "LRANGE", key, "0", "-1" }); boost::shared_ptr<redisReply> redisReply = ExecuteQuery({ "SMEMBERS", key });
VERIFY(redisReply->type == REDIS_REPLY_ARRAY); VERIFY(redisReply->type == REDIS_REPLY_ARRAY);
RedisSubscriptionInfo rsi; RedisSubscriptionInfo rsi;
Array::Ptr printer = new Array();
for (size_t j = 0; j < redisReply->elements; j++) { for (size_t j = 0; j < redisReply->elements; j++) {
rsi.EventTypes.insert(redisReply->element[j]->str); rsi.EventTypes.insert(redisReply->element[j]->str);
printer->Add(redisReply->element[j]->str);
} }
Log(LogInformation, "RedisWriter") Log(LogInformation, "RedisWriter")
<< "Subscriber Info - Key: " << key << " Value: " << printer->ToString(); << "Subscriber Info - Key: " << key << " Value: " << Value(Array::FromSet(rsi.EventTypes));
m_Subscriptions[key.SubStr(20)] = rsi;
m_Subscriptions[key.SubStr(keyPrefix.GetLength())] = rsi;
} catch (const std::exception& ex) { } catch (const std::exception& ex) {
Log(LogWarning, "RedisWriter") Log(LogWarning, "RedisWriter")
<< "Invalid Redis subscriber info for subscriber '" << key << "': " << DiagnosticInformation(ex); << "Invalid Redis subscriber info for subscriber '" << key << "': " << DiagnosticInformation(ex);
}
}
} while (cursor != 0);
continue;
}
}
Log(LogInformation, "RedisWriter") Log(LogInformation, "RedisWriter")
<< "Current Redis event subscriptions: " << m_Subscriptions.size(); << "Current Redis event subscriptions: " << m_Subscriptions.size();
} }