Skip to content

Commit 51273c2

Browse files
yhabteabyhabteab
committed
IcingaDB: put all queue related stuff into icingadb:task_queue namespace
1 parent 1a422e6 commit 51273c2

3 files changed

Lines changed: 72 additions & 66 deletions

File tree

‎lib/icingadb/icingadb-objects.cpp‎

Lines changed: 20 additions & 20 deletions
Original file line numberDiff line numberDiff line change
@@ -354,7 +354,7 @@ void IcingaDB::UpdateAllConfigObjects()
354354
auto checkable (dynamic_pointer_cast<Checkable>(object));
355355

356356
if (checkable && checkable->GetEnableActiveChecks()) {
357-
EnqueueConfigObject(checkable, NextUpdate);
357+
EnqueueConfigObject(checkable, icingadb::task_queue::NextUpdate);
358358
}
359359
}
360360

@@ -1334,15 +1334,15 @@ void IcingaDB::UpdateState(const Checkable::Ptr& checkable, uint32_t mode)
13341334
String checksum = HashValue(stateAttrs);
13351335

13361336
auto [redisStateKey, redisChecksumKey] = GetCheckableStateKeys(checkable->GetReflectionType());
1337-
if (mode & VolatileState) {
1337+
if (mode & icingadb::task_queue::VolatileState) {
13381338
String objectKey = GetObjectIdentifier(checkable);
13391339
m_RconWorker->FireAndForgetQueries({
13401340
{"HSET", redisStateKey, objectKey, JsonEncode(stateAttrs)},
13411341
{"HSET", redisChecksumKey, objectKey, JsonEncode(new Dictionary({{"checksum", checksum}}))},
13421342
});
13431343
}
13441344

1345-
if (mode & RuntimeState) {
1345+
if (mode & icingadb::task_queue::RuntimeState) {
13461346
ObjectLock olock(stateAttrs);
13471347

13481348
RedisConnection::Query streamadd({
@@ -1815,7 +1815,7 @@ void IcingaDB::SendConfigDelete(const ConfigObject::Ptr& object)
18151815

18161816
auto [configStateKey, checksumStateKey] = GetCheckableStateKeys(checkable->GetReflectionType());
18171817
EnqueueRelationsDeletion(GetObjectIdentifier(checkable), {{configStateKey, checksumStateKey}});
1818-
EnqueueConfigObject(object, ConfigDelete | NextUpdate); // Send also ZREM for next update
1818+
EnqueueConfigObject(object, icingadb::task_queue::ConfigDelete | icingadb::task_queue::NextUpdate); // Send also ZREM for next update
18191819

18201820
if (service) {
18211821
SendGroupsChanged<ServiceGroup>(checkable, service->GetGroups(), nullptr);
@@ -1826,7 +1826,7 @@ void IcingaDB::SendConfigDelete(const ConfigObject::Ptr& object)
18261826
return;
18271827
}
18281828

1829-
EnqueueConfigObject(object, ConfigDelete);
1829+
EnqueueConfigObject(object, icingadb::task_queue::ConfigDelete);
18301830

18311831
if (type == TimePeriod::TypeInstance) {
18321832
TimePeriod::Ptr timeperiod = static_pointer_cast<TimePeriod>(object);
@@ -1888,7 +1888,7 @@ void IcingaDB::SendStateChange(const ConfigObject::Ptr& object, const CheckResul
18881888

18891889
tie(host, service) = GetHostService(checkable);
18901890

1891-
EnqueueConfigObject(checkable, RuntimeState);
1891+
EnqueueConfigObject(checkable, icingadb::task_queue::RuntimeState);
18921892

18931893
int hard_state{};
18941894
if (!cr) {
@@ -2069,7 +2069,7 @@ void IcingaDB::SendStartedDowntime(const Downtime::Ptr& downtime)
20692069
return;
20702070
}
20712071

2072-
EnqueueConfigObject(downtime, ConfigUpdate);
2072+
EnqueueConfigObject(downtime, icingadb::task_queue::ConfigUpdate);
20732073

20742074
auto checkable (downtime->GetCheckable());
20752075
auto triggeredBy (Downtime::GetByName(downtime->GetTriggeredBy()));
@@ -2079,7 +2079,7 @@ void IcingaDB::SendStartedDowntime(const Downtime::Ptr& downtime)
20792079
tie(host, service) = GetHostService(checkable);
20802080

20812081
/* Update checkable state as in_downtime may have changed. */
2082-
EnqueueConfigObject(checkable, FullState);
2082+
EnqueueConfigObject(checkable, icingadb::task_queue::FullState);
20832083

20842084
RedisConnection::Query xAdd ({
20852085
"XADD", "icinga:history:stream:downtime", "*",
@@ -2168,7 +2168,7 @@ void IcingaDB::SendRemovedDowntime(const Downtime::Ptr& downtime)
21682168
return;
21692169

21702170
/* Update checkable state as in_downtime may have changed. */
2171-
EnqueueConfigObject(checkable, FullState);
2171+
EnqueueConfigObject(checkable, icingadb::task_queue::FullState);
21722172

21732173
RedisConnection::Query xAdd ({
21742174
"XADD", "icinga:history:stream:downtime", "*",
@@ -2257,7 +2257,7 @@ void IcingaDB::SendAddedComment(const Comment::Ptr& comment)
22572257
tie(host, service) = GetHostService(checkable);
22582258

22592259
// Update the checkable state to so that the "last_comment_id" is correctly reflected.
2260-
EnqueueConfigObject(checkable, FullState);
2260+
EnqueueConfigObject(checkable, icingadb::task_queue::FullState);
22612261

22622262
RedisConnection::Query xAdd ({
22632263
"XADD", "icinga:history:stream:comment", "*",
@@ -2331,7 +2331,7 @@ void IcingaDB::SendRemovedComment(const Comment::Ptr& comment)
23312331
tie(host, service) = GetHostService(checkable);
23322332

23332333
// Update the checkable state to so that the "last_comment_id" is correctly reflected.
2334-
EnqueueConfigObject(checkable, FullState);
2334+
EnqueueConfigObject(checkable, icingadb::task_queue::FullState);
23352335

23362336
RedisConnection::Query xAdd ({
23372337
"XADD", "icinga:history:stream:comment", "*",
@@ -2488,7 +2488,7 @@ void IcingaDB::SendAcknowledgementSet(const Checkable::Ptr& checkable, const Str
24882488
tie(host, service) = GetHostService(checkable);
24892489

24902490
/* Update checkable state as is_acknowledged may have changed. */
2491-
EnqueueConfigObject(checkable, FullState);
2491+
EnqueueConfigObject(checkable, icingadb::task_queue::FullState);
24922492

24932493
RedisConnection::Query xAdd ({
24942494
"XADD", "icinga:history:stream:acknowledgement", "*",
@@ -2546,7 +2546,7 @@ void IcingaDB::SendAcknowledgementCleared(const Checkable::Ptr& checkable, const
25462546
tie(host, service) = GetHostService(checkable);
25472547

25482548
/* Update checkable state as is_acknowledged may have changed. */
2549-
EnqueueConfigObject(checkable, FullState);
2549+
EnqueueConfigObject(checkable, icingadb::task_queue::FullState);
25502550

25512551
RedisConnection::Query xAdd ({
25522552
"XADD", "icinga:history:stream:acknowledgement", "*",
@@ -2965,7 +2965,7 @@ void IcingaDB::ReachabilityChangeHandler(const std::set<Checkable::Ptr>& childre
29652965
{
29662966
for (const IcingaDB::Ptr& rw : ConfigType::GetObjectsByType<IcingaDB>()) {
29672967
for (auto& checkable : children) {
2968-
rw->EnqueueConfigObject(checkable, FullState);
2968+
rw->EnqueueConfigObject(checkable, icingadb::task_queue::FullState);
29692969
for (const auto& dependencyGroup : checkable->GetDependencyGroups()) {
29702970
rw->EnqueueDependencyGroupStateUpdate(dependencyGroup);
29712971
}
@@ -2986,7 +2986,7 @@ void IcingaDB::VersionChangedHandler(const ConfigObject::Ptr& object)
29862986
if (object->IsActive()) {
29872987
for (const IcingaDB::Ptr& rw : ConfigType::GetObjectsByType<IcingaDB>()) {
29882988
// A runtime config change triggers also a full state update as well as next update event.
2989-
rw->EnqueueConfigObject(object, ConfigUpdate | FullState | NextUpdate);
2989+
rw->EnqueueConfigObject(object, icingadb::task_queue::ConfigUpdate | icingadb::task_queue::FullState | icingadb::task_queue::NextUpdate);
29902990
}
29912991
} else if (!object->IsActive() && object->GetExtension("ConfigObjectDeleted")) { // same as in apilistener-configsync.cpp
29922992
for (const IcingaDB::Ptr& rw : ConfigType::GetObjectsByType<IcingaDB>()) {
@@ -3050,21 +3050,21 @@ void IcingaDB::FlappingChangeHandler(const Checkable::Ptr& checkable, double cha
30503050
void IcingaDB::NewCheckResultHandler(const Checkable::Ptr& checkable)
30513051
{
30523052
for (auto& rw : ConfigType::GetObjectsByType<IcingaDB>()) {
3053-
rw->EnqueueConfigObject(checkable, VolatileState);
3053+
rw->EnqueueConfigObject(checkable, icingadb::task_queue::VolatileState);
30543054
}
30553055
}
30563056

30573057
void IcingaDB::NextCheckChangedHandler(const Checkable::Ptr& checkable)
30583058
{
30593059
for (auto& rw : ConfigType::GetObjectsByType<IcingaDB>()) {
3060-
rw->EnqueueConfigObject(checkable, VolatileState | NextUpdate);
3060+
rw->EnqueueConfigObject(checkable, icingadb::task_queue::VolatileState | icingadb::task_queue::NextUpdate);
30613061
}
30623062
}
30633063

30643064
void IcingaDB::DependencyGroupChildRegisteredHandler(const Checkable::Ptr& child, const DependencyGroup::Ptr& dependencyGroup)
30653065
{
30663066
for (const auto& rw : ConfigType::GetObjectsByType<IcingaDB>()) {
3067-
rw->EnqueueConfigObject(child, FullState); // Child requires a full state update.
3067+
rw->EnqueueConfigObject(child, icingadb::task_queue::FullState); // Child requires a full state update.
30683068
rw->EnqueueDependencyChildRegistered(dependencyGroup, child);
30693069
rw->EnqueueDependencyGroupStateUpdate(dependencyGroup);
30703070

@@ -3075,7 +3075,7 @@ void IcingaDB::DependencyGroupChildRegisteredHandler(const Checkable::Ptr& child
30753075
// Checkable as well. The grandparent Checkable may still have wrong numbers of total children, though it's
30763076
// not worth traversing the whole tree way up and sending config updates for each one of them, as the next
30773077
// Redis config dump is going to fix it anyway.
3078-
rw->EnqueueConfigObject(parent, ConfigUpdate | FullState);
3078+
rw->EnqueueConfigObject(parent, icingadb::task_queue::ConfigUpdate | icingadb::task_queue::FullState);
30793079
}
30803080
}
30813081
}
@@ -3090,7 +3090,7 @@ void IcingaDB::DependencyGroupChildRemovedHandler(const DependencyGroup::Ptr& de
30903090
void IcingaDB::HostProblemChangedHandler(const Service::Ptr& service) {
30913091
for (auto& rw : ConfigType::GetObjectsByType<IcingaDB>()) {
30923092
/* Host state changes affect is_handled and severity of services. */
3093-
rw->EnqueueConfigObject(service, FullState);
3093+
rw->EnqueueConfigObject(service, icingadb::task_queue::FullState);
30943094
}
30953095
}
30963096

‎lib/icingadb/icingadb-worker.cpp‎

Lines changed: 27 additions & 26 deletions
Original file line numberDiff line numberDiff line change
@@ -7,22 +7,22 @@
77

88
using namespace icinga;
99

10-
PendingConfigItem::PendingConfigItem(const ConfigObject::Ptr& obj, uint32_t bits)
10+
icingadb::task_queue::PendingConfigItem::PendingConfigItem(const ConfigObject::Ptr& obj, uint32_t bits)
1111
: Object{obj}, DirtyBits{bits & DirtyBitsAll}
1212
{
1313
}
1414

15-
PendingDependencyGroupStateItem::PendingDependencyGroupStateItem(const DependencyGroup::Ptr& depGroup)
15+
icingadb::task_queue::PendingDependencyGroupStateItem::PendingDependencyGroupStateItem(const DependencyGroup::Ptr& depGroup)
1616
: DepGroup{depGroup}
1717
{
1818
}
1919

20-
PendingDependencyEdgeItem::PendingDependencyEdgeItem(const DependencyGroup::Ptr& depGroup, const Checkable::Ptr& child)
20+
icingadb::task_queue::PendingDependencyEdgeItem::PendingDependencyEdgeItem(const DependencyGroup::Ptr& depGroup, const Checkable::Ptr& child)
2121
: DepGroup{depGroup}, Child{child}
2222
{
2323
}
2424

25-
RelationsDeletionItem::RelationsDeletionItem(const String& id, const RelationsKeySet& relations)
25+
icingadb::task_queue::RelationsDeletionItem::RelationsDeletionItem(const String& id, const RelationsKeySet& relations)
2626
: ID{id}, Relations{relations}
2727
{
2828
}
@@ -93,7 +93,7 @@ std::chrono::duration<double> IcingaDB::DequeueAndProcessOne(std::unique_lock<st
9393
auto& seqView = m_PendingItems.get<1>();
9494
for (auto it(seqView.begin()); it != seqView.end(); ++it) {
9595
if (it != seqView.begin()) {
96-
if (dynamic_cast<const RelationsDeletionItem*>(it->get())) {
96+
if (dynamic_cast<const icingadb::task_queue::RelationsDeletionItem*>(it->get())) {
9797
// We don't know whether the previous items are related to this deletion item or not,
9898
// thus we can't just process this right now when there are older items in the queue.
9999
// Otherwise, we might delete something that is going to be updated/created.
@@ -122,7 +122,7 @@ std::chrono::duration<double> IcingaDB::DequeueAndProcessOne(std::unique_lock<st
122122
try {
123123
itemToProcess->Execute(*this);
124124
} catch (const std::exception& ex) {
125-
PendingQueueItem& itemRef = *itemToProcess; // For typeid(operand of typeid must not have any side effects).
125+
icingadb::task_queue::PendingQueueItem& itemRef = *itemToProcess; // For typeid(operand of typeid must not have any side effects).
126126
Log(LogCritical, "IcingaDB")
127127
<< "Exception while processing pending item of type '" << typeid(itemRef).name() << "': "
128128
<< DiagnosticInformation(ex, GetActive());
@@ -138,7 +138,7 @@ std::chrono::duration<double> IcingaDB::DequeueAndProcessOne(std::unique_lock<st
138138
return retryAfter;
139139
}
140140

141-
ConfigObject::Ptr PendingConfigItem::GetObjectToLock() const
141+
ConfigObject::Ptr icingadb::task_queue::PendingConfigItem::GetObjectToLock() const
142142
{
143143
return Object;
144144
}
@@ -152,7 +152,7 @@ ConfigObject::Ptr PendingConfigItem::GetObjectToLock() const
152152
*
153153
* @param icingadb The IcingaDB instance to use for executing Redis queries.
154154
*/
155-
void PendingConfigItem::Execute(IcingaDB& icingadb) const {
155+
void icingadb::task_queue::PendingConfigItem::Execute(IcingaDB& icingadb) const {
156156
if (DirtyBits & ConfigDelete) {
157157
auto redisKeyPair = icingadb.GetSyncableTypeRedisKeys(Object->GetReflectionType());
158158
icingadb.m_RconWorker->FireAndForgetQueries(
@@ -203,7 +203,7 @@ void PendingConfigItem::Execute(IcingaDB& icingadb) const {
203203
*
204204
* @param icingadb The IcingaDB instance to use for executing Redis queries.
205205
*/
206-
void PendingDependencyGroupStateItem::Execute(IcingaDB& icingadb) const
206+
void icingadb::task_queue::PendingDependencyGroupStateItem::Execute(IcingaDB& icingadb) const
207207
{
208208
// For dependency group state updates, we don't actually care which child triggered the update,
209209
// since all children share the same dependency group state. Thus, we can just pick any child to
@@ -221,7 +221,7 @@ void PendingDependencyGroupStateItem::Execute(IcingaDB& icingadb) const
221221
*
222222
* @param icingadb The IcingaDB instance to use for executing Redis queries.
223223
*/
224-
void PendingDependencyEdgeItem::Execute(IcingaDB& icingadb) const
224+
void icingadb::task_queue::PendingDependencyEdgeItem::Execute(IcingaDB& icingadb) const
225225
{
226226
std::vector<Dictionary::Ptr> runtimeUpdates;
227227
std::map<RedisConnection::QueryArg, RedisConnection::Query> hMSets;
@@ -238,7 +238,7 @@ void PendingDependencyEdgeItem::Execute(IcingaDB& icingadb) const
238238
*
239239
* @param icingadb The IcingaDB instance to use for executing Redis queries.
240240
*/
241-
void RelationsDeletionItem::Execute(IcingaDB& icingadb) const
241+
void icingadb::task_queue::RelationsDeletionItem::Execute(IcingaDB& icingadb) const
242242
{
243243
for (const auto& [configKey, checksumKey] : Relations) {
244244
if (icingadb.IsStateKey(configKey)) {
@@ -260,18 +260,19 @@ void IcingaDB::EnqueueConfigObject(const ConfigObject::Ptr& object, uint32_t bit
260260
if (!GetActive() || !m_RconWorker || !m_RconWorker->IsConnected()) {
261261
return; // No need to enqueue anything if we're not connected.
262262
}
263+
namespace queue = icingadb::task_queue;
263264

264265
{
265266
std::lock_guard lock(m_PendingItemsMutex);
266-
if (auto [it, inserted] = m_PendingItems.insert(std::make_shared<PendingConfigItem>(object, bits)); !inserted) {
267-
m_PendingItems.modify(it, [bits](const std::shared_ptr<PendingQueueItem>& item) {
268-
auto configItem = dynamic_cast<PendingConfigItem*>(item.get());
269-
if (bits & ConfigDelete) {
270-
configItem->DirtyBits &= ~(ConfigUpdate | FullState);
271-
} else if (bits & ConfigUpdate) {
272-
configItem->DirtyBits &= ~ConfigDelete;
267+
if (auto [it, inserted] = m_PendingItems.insert(std::make_shared<queue::PendingConfigItem>(object, bits)); !inserted) {
268+
m_PendingItems.modify(it, [bits](const std::shared_ptr<queue::PendingQueueItem>& item) {
269+
auto configItem = dynamic_cast<queue::PendingConfigItem*>(item.get());
270+
if (bits & queue::ConfigDelete) {
271+
configItem->DirtyBits &= ~(queue::ConfigUpdate | queue::FullState);
272+
} else if (bits & queue::ConfigUpdate) {
273+
configItem->DirtyBits &= ~queue::ConfigDelete;
273274
}
274-
configItem->DirtyBits |= bits & DirtyBitsAll;
275+
configItem->DirtyBits |= bits & queue::DirtyBitsAll;
275276
});
276277
}
277278
}
@@ -283,7 +284,7 @@ void IcingaDB::EnqueueDependencyGroupStateUpdate(const DependencyGroup::Ptr& dep
283284
if (GetActive() && m_RconWorker && m_RconWorker->IsConnected()) {
284285
{
285286
std::lock_guard lock(m_PendingItemsMutex);
286-
m_PendingItems.insert(std::make_shared<PendingDependencyGroupStateItem>(depGroup));
287+
m_PendingItems.insert(std::make_shared<icingadb::task_queue::PendingDependencyGroupStateItem>(depGroup));
287288
}
288289
m_PendingItemsCV.notify_one();
289290
}
@@ -303,7 +304,7 @@ void IcingaDB::EnqueueDependencyChildRegistered(const DependencyGroup::Ptr& depG
303304
if (GetActive() && m_RconWorker && m_RconWorker->IsConnected()) {
304305
{
305306
std::lock_guard lock(m_PendingItemsMutex);
306-
m_PendingItems.insert(std::make_shared<PendingDependencyEdgeItem>(depGroup, child));
307+
m_PendingItems.insert(std::make_shared<icingadb::task_queue::PendingDependencyEdgeItem>(depGroup, child));
307308
}
308309
m_PendingItemsCV.notify_one();
309310
}
@@ -409,7 +410,7 @@ void IcingaDB::EnqueueDependencyChildRemoved(
409410
// Checkable as well. The grandparent Checkable may still have wrong numbers of total children, though it's
410411
// not worth traversing the whole tree way up and sending config updates for each one of them, as the next
411412
// Redis config dump is going to fix it anyway.
412-
EnqueueConfigObject(parent, ConfigUpdate);
413+
EnqueueConfigObject(parent, icingadb::task_queue::ConfigUpdate);
413414

414415
if (!parent->HasAnyDependencies()) {
415416
// If the parent Checkable isn't part of any other dependency chain anymore, drop its dependency node entry.
@@ -446,17 +447,17 @@ void IcingaDB::EnqueueDependencyChildRemoved(
446447
* @param id The ID of the relation to be deleted.
447448
* @param relations A map of Redis keys from which to delete the relation.
448449
*/
449-
void IcingaDB::EnqueueRelationsDeletion(const String& id, const RelationsDeletionItem::RelationsKeySet& relations)
450+
void IcingaDB::EnqueueRelationsDeletion(const String& id, const icingadb::task_queue::RelationsDeletionItem::RelationsKeySet& relations)
450451
{
451452
if (!GetActive() || !m_RconWorker || !m_RconWorker->IsConnected()) {
452453
return; // No need to enqueue anything if we're not connected.
453454
}
454455

455456
{
456457
std::lock_guard lock(m_PendingItemsMutex);
457-
if (auto [it, inserted] = m_PendingItems.insert(std::make_shared<RelationsDeletionItem>(id, relations)); !inserted) {
458-
m_PendingItems.modify(it, [&relations](std::shared_ptr<PendingQueueItem>& val) {
459-
auto item = dynamic_cast<RelationsDeletionItem*>(val.get());
458+
if (auto [it, inserted] = m_PendingItems.insert(std::make_shared<icingadb::task_queue::RelationsDeletionItem>(id, relations)); !inserted) {
459+
m_PendingItems.modify(it, [&relations](std::shared_ptr<icingadb::task_queue::PendingQueueItem>& val) {
460+
auto item = dynamic_cast<icingadb::task_queue::RelationsDeletionItem*>(val.get());
460461
item->Relations.insert(relations.begin(), relations.end());
461462
});
462463
}

0 commit comments

Comments
 (0)