IcingaDB: include consistent IDs, start time and end time in icinga:history:stream:{flapping,acknowledgement}

refs #7689

lib/icinga/checkable-flapping.cpp

@@ -64,10 +64,16 @@ void Checkable::UpdateFlappingStatus(bool stateChange)
 	SetFlappingBuffer(stateChangeBuf.GetValue());
 	SetFlappingIndex(oldestIndex);
 	SetFlappingCurrent(flappingValue);
-	SetFlapping(flapping, true);
 
-	if (flapping != GetFlapping())
+	if (flapping != GetFlapping()) {
-		SetFlappingLastChange(GetLastCheckResult()->GetExecutionEnd());
+		SetFlapping(flapping, true);
+
+		double ee = GetLastCheckResult()->GetExecutionEnd();
+
+		OnFlappingChange(this, ee);
+
+		SetFlappingLastChange(ee);
+	}
 }
 
 bool Checkable::IsFlapping() const

lib/icinga/checkable.cpp

@@ -17,6 +17,7 @@ INITIALIZE_ONCE(&Checkable::StaticInitialize);
 
 boost::signals2::signal<void (const Checkable::Ptr&, const String&, const String&, AcknowledgementType, bool, bool, double, double, const MessageOrigin::Ptr&)> Checkable::OnAcknowledgementSet;
 boost::signals2::signal<void (const Checkable::Ptr&, const String&, double, const MessageOrigin::Ptr&)> Checkable::OnAcknowledgementCleared;
+boost::signals2::signal<void (const Checkable::Ptr&, double)> Checkable::OnFlappingChange;
 
 static Timer::Ptr l_CheckablesFireSuppressedNotifications;
 

lib/icinga/checkable.hpp

@@ -114,6 +114,7 @@ public:
 	static boost::signals2::signal<void (const Checkable::Ptr&, const String&, const String&, AcknowledgementType,
 		bool, bool, double, double, const MessageOrigin::Ptr&)> OnAcknowledgementSet;
 	static boost::signals2::signal<void (const Checkable::Ptr&, const String&, double, const MessageOrigin::Ptr&)> OnAcknowledgementCleared;
+	static boost::signals2::signal<void (const Checkable::Ptr&, double)> OnFlappingChange;
 	static boost::signals2::signal<void (const Checkable::Ptr&)> OnNextCheckUpdated;
 	static boost::signals2::signal<void (const Checkable::Ptr&)> OnEventCommandExecuted;
 

lib/icingadb/icingadb-objects.cpp

@@ -96,7 +96,7 @@ void IcingaDB::ConfigStaticInitialize()
 	Comment::OnCommentAdded.connect(&IcingaDB::CommentAddedHandler);
 	Comment::OnCommentRemoved.connect(&IcingaDB::CommentRemovedHandler);
 
-	Checkable::OnFlappingChanged.connect(&IcingaDB::FlappingChangedHandler);
+	Checkable::OnFlappingChange.connect(&IcingaDB::FlappingChangeHandler);
 
 	Checkable::OnNewCheckResult.connect([](const Checkable::Ptr& checkable, const CheckResult::Ptr&, const MessageOrigin::Ptr&) {
 		IcingaDB::NewCheckResultHandler(checkable);
@@ -1611,7 +1611,7 @@ void IcingaDB::SendRemovedComment(const Comment::Ptr& comment)
 	m_Rcon->FireAndForgetQuery(std::move(xAdd), Prio::History);
 }
 
-void IcingaDB::SendFlappingChanged(const Checkable::Ptr& checkable, const Value& value)
+void IcingaDB::SendFlappingChange(const Checkable::Ptr& checkable, double changeTime, double flappingLastChange)
 {
 	if (!m_Rcon || !m_Rcon->IsConnected())
 		return;
@@ -1622,15 +1622,12 @@ void IcingaDB::SendFlappingChanged(const Checkable::Ptr& checkable, const Value&
 
 	std::vector<String> xAdd ({
 		"XADD", "icinga:history:stream:flapping", "*",
-		"id", Utility::NewUniqueID(),
 		"environment_id", SHA1(GetEnvironment()),
 		"host_id", GetObjectIdentifier(host),
 		"percent_state_change", Convert::ToString(checkable->GetFlappingCurrent()),
 		"flapping_threshold_low", Convert::ToString(checkable->GetFlappingThresholdLow()),
 		"flapping_threshold_high", Convert::ToString(checkable->GetFlappingThresholdHigh()),
-		"event_time", Convert::ToString(TimestampToMilliseconds(Utility::GetTime())),
+		"event_id", Utility::NewUniqueID()
-		"event_id", Utility::NewUniqueID(),
-		"event_type", value.ToBool() ? "flapping_start" : "flapping_end"
 	});
 
 	if (service) {
@@ -1650,6 +1647,27 @@ void IcingaDB::SendFlappingChanged(const Checkable::Ptr& checkable, const Value&
 		xAdd.emplace_back(GetObjectIdentifier(endpoint));
 	}
 
+	long long startTime;
+
+	if (checkable->IsFlapping()) {
+		startTime = TimestampToMilliseconds(changeTime);
+
+		xAdd.emplace_back("event_type");
+		xAdd.emplace_back("flapping_start");
+	} else {
+		startTime = TimestampToMilliseconds(flappingLastChange);
+
+		xAdd.emplace_back("event_type");
+		xAdd.emplace_back("flapping_end");
+		xAdd.emplace_back("end_time");
+		xAdd.emplace_back(Convert::ToString(TimestampToMilliseconds(changeTime)));
+	}
+
+	xAdd.emplace_back("start_time");
+	xAdd.emplace_back(Convert::ToString(startTime));
+	xAdd.emplace_back("id");
+	xAdd.emplace_back(HashValue(new Array({GetEnvironment(), checkable->GetReflectionType()->GetName(), checkable->GetName(), startTime})));
+
 	m_Rcon->FireAndForgetQuery(std::move(xAdd), Prio::History);
 }
 
@@ -1680,10 +1698,9 @@ void IcingaDB::SendAcknowledgementSet(const Checkable::Ptr& checkable, const Str
 
 	std::vector<String> xAdd ({
 		"XADD", "icinga:history:stream:acknowledgement", "*",
-		"id", Utility::NewUniqueID(),
+		"event_id", Utility::NewUniqueID(),
 		"environment_id", SHA1(GetEnvironment()),
 		"host_id", GetObjectIdentifier(host),
-		"event_time", Convert::ToString(TimestampToMilliseconds(changeTime)),
 		"event_type", "ack_set",
 		"author", author,
 		"comment", comment,
@@ -1713,10 +1730,17 @@ void IcingaDB::SendAcknowledgementSet(const Checkable::Ptr& checkable, const Str
 		xAdd.emplace_back(Convert::ToString(TimestampToMilliseconds(expiry)));
 	}
 
+	long long setTime = TimestampToMilliseconds(changeTime);
+
+	xAdd.emplace_back("set_time");
+	xAdd.emplace_back(Convert::ToString(setTime));
+	xAdd.emplace_back("id");
+	xAdd.emplace_back(HashValue(new Array({GetEnvironment(), checkable->GetReflectionType()->GetName(), checkable->GetName(), setTime})));
+
 	m_Rcon->FireAndForgetQuery(std::move(xAdd), Prio::History);
 }
 
-void IcingaDB::SendAcknowledgementCleared(const Checkable::Ptr& checkable, const String& removedBy, double changeTime)
+void IcingaDB::SendAcknowledgementCleared(const Checkable::Ptr& checkable, const String& removedBy, double changeTime, double ackLastChange)
 {
 	if (!m_Rcon || !m_Rcon->IsConnected())
 		return;
@@ -1727,12 +1751,11 @@ void IcingaDB::SendAcknowledgementCleared(const Checkable::Ptr& checkable, const
 
 	std::vector<String> xAdd ({
 		"XADD", "icinga:history:stream:acknowledgement", "*",
-		"id", Utility::NewUniqueID(),
+		"event_id", Utility::NewUniqueID(),
 		"environment_id", SHA1(GetEnvironment()),
 		"host_id", GetObjectIdentifier(host),
-		"event_time", Convert::ToString(TimestampToMilliseconds(changeTime)),
+		"clear_time", Convert::ToString(TimestampToMilliseconds(changeTime)),
-		"event_type", "ack_clear",
+		"event_type", "ack_clear"
-		"author", removedBy
 	});
 
 	if (service) {
@@ -1752,6 +1775,18 @@ void IcingaDB::SendAcknowledgementCleared(const Checkable::Ptr& checkable, const
 		xAdd.emplace_back(GetObjectIdentifier(endpoint));
 	}
 
+	long long setTime = TimestampToMilliseconds(ackLastChange);
+
+	xAdd.emplace_back("set_time");
+	xAdd.emplace_back(Convert::ToString(setTime));
+	xAdd.emplace_back("id");
+	xAdd.emplace_back(HashValue(new Array({GetEnvironment(), checkable->GetReflectionType()->GetName(), checkable->GetName(), setTime})));
+
+	if (!removedBy.IsEmpty()) {
+		xAdd.emplace_back("cleared_by");
+		xAdd.emplace_back(removedBy);
+	}
+
 	m_Rcon->FireAndForgetQuery(std::move(xAdd), Prio::History);
 }
 
@@ -1987,10 +2022,12 @@ void IcingaDB::CommentRemovedHandler(const Comment::Ptr& comment)
 	}
 }
 
-void IcingaDB::FlappingChangedHandler(const Checkable::Ptr& checkable, const Value& value)
+void IcingaDB::FlappingChangeHandler(const Checkable::Ptr& checkable, double changeTime)
 {
+	auto flappingLastChange (checkable->GetFlappingLastChange());
+
 	for (auto& rw : ConfigType::GetObjectsByType<IcingaDB>()) {
-		rw->m_WorkQueue.Enqueue([rw, checkable, value]() { rw->SendFlappingChanged(checkable, value); });
+		rw->m_WorkQueue.Enqueue([rw, checkable, changeTime, flappingLastChange]() { rw->SendFlappingChange(checkable, changeTime, flappingLastChange); });
 	}
 }
 
@@ -2026,9 +2063,10 @@ void IcingaDB::AcknowledgementClearedHandler(const Checkable::Ptr& checkable, co
 
 	if (!rws.empty()) {
 		auto rb (Shared<String>::Make(removedBy));
+		auto ackLastChange (checkable->GetAcknowledgementLastChange());
 
 		for (auto& rw : rws) {
-			rw->m_WorkQueue.Enqueue([rw, checkable, rb, changeTime]() { rw->SendAcknowledgementCleared(checkable, *rb, changeTime); });
+			rw->m_WorkQueue.Enqueue([rw, checkable, rb, changeTime, ackLastChange]() { rw->SendAcknowledgementCleared(checkable, *rb, changeTime, ackLastChange); });
 		}
 	}
 }

lib/icingadb/icingadb.hpp

@@ -65,10 +65,10 @@ private:
 	void SendRemovedDowntime(const Downtime::Ptr& downtime);
 	void SendAddedComment(const Comment::Ptr& comment);
 	void SendRemovedComment(const Comment::Ptr& comment);
-	void SendFlappingChanged(const Checkable::Ptr& checkable, const Value& value);
+	void SendFlappingChange(const Checkable::Ptr& checkable, double changeTime, double flappingLastChange);
 	void SendNextUpdate(const Checkable::Ptr& checkable);
 	void SendAcknowledgementSet(const Checkable::Ptr& checkable, const String& author, const String& comment, AcknowledgementType type, bool persistent, double changeTime, double expiry);
-	void SendAcknowledgementCleared(const Checkable::Ptr& checkable, const String& removedBy, double changeTime);
+	void SendAcknowledgementCleared(const Checkable::Ptr& checkable, const String& removedBy, double changeTime, double ackLastChange);
 
 	std::vector<String> UpdateObjectAttrs(const ConfigObject::Ptr& object, int fieldType, const String& typeNameOverride);
 	Dictionary::Ptr SerializeState(const Checkable::Ptr& checkable);
@@ -104,7 +104,7 @@ private:
 
 	static void CommentAddedHandler(const Comment::Ptr& comment);
 	static void CommentRemovedHandler(const Comment::Ptr& comment);
-	static void FlappingChangedHandler(const Checkable::Ptr& checkable, const Value& value);
+	static void FlappingChangeHandler(const Checkable::Ptr& checkable, double changeTime);
 	static void NewCheckResultHandler(const Checkable::Ptr& checkable);
 	static void AcknowledgementSetHandler(const Checkable::Ptr& checkable, const String& author, const String& comment, AcknowledgementType type, bool persistent, double changeTime, double expiry);
 	static void AcknowledgementClearedHandler(const Checkable::Ptr& checkable, const String& removedBy, double changeTime);