Merge pull request #5503 from Icinga/fix/livestatus-stats-groups

Fix grouping for Livestatus queries with 'Stats'

lib/livestatus/aggregator.cpp

@@ -33,3 +33,6 @@ Filter::Ptr Aggregator::GetFilter(void) const
 {
 	return m_Filter;
 }
+
+AggregatorState::~AggregatorState(void)
+{ }

lib/livestatus/aggregator.hpp

@@ -27,6 +27,14 @@
 namespace icinga
 {
 
+/**
+ * @ingroup livestatus
+ */
+struct I2_LIVESTATUS_API AggregatorState
+{
+	virtual ~AggregatorState(void);
+};
+
 /**
  * @ingroup livestatus
  */
@@ -35,8 +43,8 @@ class I2_LIVESTATUS_API Aggregator : public Object
 public:
 	DECLARE_PTR_TYPEDEFS(Aggregator);
 
-	virtual void Apply(const Table::Ptr& table, const Value& row) = 0;
+	virtual void Apply(const Table::Ptr& table, const Value& row, AggregatorState **state) = 0;
-	virtual double GetResult(void) const = 0;
+	virtual double GetResultAndFreeState(AggregatorState *state) const = 0;
 	void SetFilter(const Filter::Ptr& filter);
 
 protected:

lib/livestatus/avgaggregator.cpp

@@ -22,20 +22,34 @@
 using namespace icinga;
 
 AvgAggregator::AvgAggregator(const String& attr)
-    : m_Avg(0), m_AvgCount(0), m_AvgAttr(attr)
+    : m_AvgAttr(attr)
 { }
 
-void AvgAggregator::Apply(const Table::Ptr& table, const Value& row)
+AvgAggregatorState *AvgAggregator::EnsureState(AggregatorState **state)
+{
+	if (!*state)
+		*state = new AvgAggregatorState();
+
+	return static_cast<AvgAggregatorState *>(*state);
+}
+
+void AvgAggregator::Apply(const Table::Ptr& table, const Value& row, AggregatorState **state)
 {
 	Column column = table->GetColumn(m_AvgAttr);
 
 	Value value = column.ExtractValue(row);
 
-	m_Avg += value;
+	AvgAggregatorState *pstate = EnsureState(state);
-	m_AvgCount++;
+
+	pstate->Avg += value;
+	pstate->AvgCount++;
 }
 
-double AvgAggregator::GetResult(void) const
+double AvgAggregator::GetResultAndFreeState(AggregatorState *state) const
 {
-	return (m_Avg / m_AvgCount);
+	AvgAggregatorState *pstate = EnsureState(&state);
+	double result = pstate->Avg / pstate->AvgCount;
+	delete pstate;
+
+	return result;
 }

lib/livestatus/avgaggregator.hpp

@@ -26,6 +26,19 @@
 namespace icinga
 {
 
+/**
+ * @ingroup livestatus
+ */
+struct AvgAggregatorState : public AggregatorState
+{
+	AvgAggregatorState(void)
+	    : Avg(0), AvgCount(0)
+	{ }
+
+	double Avg;
+	double AvgCount;
+};
+
 /**
  * @ingroup livestatus
  */
@@ -36,13 +49,13 @@ public:
 
 	AvgAggregator(const String& attr);
 
-	virtual void Apply(const Table::Ptr& table, const Value& row) override;
+	virtual void Apply(const Table::Ptr& table, const Value& row, AggregatorState **state) override;
-	virtual double GetResult(void) const override;
+	virtual double GetResultAndFreeState(AggregatorState *state) const override;
 
 private:
-	double m_Avg;
-	double m_AvgCount;
 	String m_AvgAttr;
+
+	static AvgAggregatorState *EnsureState(AggregatorState **state);
 };
 
 }

lib/livestatus/countaggregator.cpp

@@ -21,17 +21,27 @@
 
 using namespace icinga;
 
-CountAggregator::CountAggregator(void)
+CountAggregatorState *CountAggregator::EnsureState(AggregatorState **state)
-    : m_Count(0)
-{ }
-
-void CountAggregator::Apply(const Table::Ptr& table, const Value& row)
 {
+	if (!*state)
+		*state = new CountAggregatorState();
+
+	return static_cast<CountAggregatorState *>(*state);
+}
+
+void CountAggregator::Apply(const Table::Ptr& table, const Value& row, AggregatorState **state)
+{
+	CountAggregatorState *pstate = EnsureState(state);
+
 	if (GetFilter()->Apply(table, row))
-		m_Count++;
+		pstate->Count++;
 }
 
-double CountAggregator::GetResult(void) const
+double CountAggregator::GetResultAndFreeState(AggregatorState *state) const
 {
-	return m_Count;
+	CountAggregatorState *pstate = EnsureState(&state);
+	double result = pstate->Count;
+	delete pstate;
+
+	return result;
 }

lib/livestatus/countaggregator.hpp

@@ -26,6 +26,18 @@
 namespace icinga
 {
 
+/**
+ * @ingroup livestatus
+ */
+struct CountAggregatorState : public AggregatorState
+{
+	CountAggregatorState(void)
+	    : Count(0)
+	{ }
+
+	int Count;
+};
+
 /**
  * @ingroup livestatus
  */
@@ -34,13 +46,11 @@ class I2_LIVESTATUS_API CountAggregator : public Aggregator
 public:
 	DECLARE_PTR_TYPEDEFS(CountAggregator);
 
-	CountAggregator(void);
+	virtual void Apply(const Table::Ptr& table, const Value& row, AggregatorState **) override;
-
+	virtual double GetResultAndFreeState(AggregatorState *state) const override;
-	virtual void Apply(const Table::Ptr& table, const Value& row) override;
-	virtual double GetResult(void) const override;
 
 private:
-	int m_Count;
+	static CountAggregatorState *EnsureState(AggregatorState **state);
 };
 
 }

lib/livestatus/invavgaggregator.cpp

@@ -22,20 +22,34 @@
 using namespace icinga;
 
 InvAvgAggregator::InvAvgAggregator(const String& attr)
-    : m_InvAvg(0), m_InvAvgCount(0), m_InvAvgAttr(attr)
+    : m_InvAvgAttr(attr)
 { }
 
-void InvAvgAggregator::Apply(const Table::Ptr& table, const Value& row)
+InvAvgAggregatorState *InvAvgAggregator::EnsureState(AggregatorState **state)
+{
+	if (!*state)
+		*state = new InvAvgAggregatorState();
+
+	return static_cast<InvAvgAggregatorState *>(*state);
+}
+
+void InvAvgAggregator::Apply(const Table::Ptr& table, const Value& row, AggregatorState **state)
 {
 	Column column = table->GetColumn(m_InvAvgAttr);
 
 	Value value = column.ExtractValue(row);
 
-	m_InvAvg += (1.0 / value);
+	InvAvgAggregatorState *pstate = EnsureState(state);
-	m_InvAvgCount++;
+
+	pstate->InvAvg += (1.0 / value);
+	pstate->InvAvgCount++;
 }
 
-double InvAvgAggregator::GetResult(void) const
+double InvAvgAggregator::GetResultAndFreeState(AggregatorState *state) const
 {
-	return (m_InvAvg / m_InvAvgCount);
+	InvAvgAggregatorState *pstate = EnsureState(&state);
+	double result = pstate->InvAvg / pstate->InvAvgCount;
+	delete pstate;
+
+	return result;
 }

lib/livestatus/invavgaggregator.hpp

@@ -26,6 +26,19 @@
 namespace icinga
 {
 
+/**
+ * @ingroup livestatus
+ */
+struct InvAvgAggregatorState : public AggregatorState
+{
+	InvAvgAggregatorState(void)
+	    : InvAvg(0), InvAvgCount(0)
+	{ }
+
+	double InvAvg;
+	double InvAvgCount;
+};
+
 /**
  * @ingroup livestatus
  */
@@ -36,13 +49,13 @@ public:
 
 	InvAvgAggregator(const String& attr);
 
-	virtual void Apply(const Table::Ptr& table, const Value& row) override;
+	virtual void Apply(const Table::Ptr& table, const Value& row, AggregatorState **state) override;
-	virtual double GetResult(void) const override;
+	virtual double GetResultAndFreeState(AggregatorState *state) const override;
 
 private:
-	double m_InvAvg;
-	double m_InvAvgCount;
 	String m_InvAvgAttr;
+
+	static InvAvgAggregatorState *EnsureState(AggregatorState **state);
 };
 
 }

lib/livestatus/invsumaggregator.cpp

@@ -22,19 +22,33 @@
 using namespace icinga;
 
 InvSumAggregator::InvSumAggregator(const String& attr)
-    : m_InvSum(0), m_InvSumAttr(attr)
+    : m_InvSumAttr(attr)
 { }
 
-void InvSumAggregator::Apply(const Table::Ptr& table, const Value& row)
+InvSumAggregatorState *InvSumAggregator::EnsureState(AggregatorState **state)
+{
+	if (!*state)
+		*state = new InvSumAggregatorState();
+
+	return static_cast<InvSumAggregatorState *>(*state);
+}
+
+void InvSumAggregator::Apply(const Table::Ptr& table, const Value& row, AggregatorState **state)
 {
 	Column column = table->GetColumn(m_InvSumAttr);
 
 	Value value = column.ExtractValue(row);
 
-	m_InvSum += (1.0 / value);
+	InvSumAggregatorState *pstate = EnsureState(state);
+
+	pstate->InvSum += (1.0 / value);
 }
 
-double InvSumAggregator::GetResult(void) const
+double InvSumAggregator::GetResultAndFreeState(AggregatorState *state) const
 {
-	return m_InvSum;
+	InvSumAggregatorState *pstate = EnsureState(&state);
+	double result = pstate->InvSum;
+	delete pstate;
+
+	return result;
 }

lib/livestatus/invsumaggregator.hpp

@@ -26,6 +26,18 @@
 namespace icinga
 {
 
+/**
+ * @ingroup livestatus
+ */
+struct I2_LIVESTATUS_API InvSumAggregatorState : public AggregatorState
+{
+	InvSumAggregatorState(void)
+	    : InvSum(0)
+	{ }
+
+	double InvSum;
+};
+
 /**
  * @ingroup livestatus
  */
@@ -36,12 +48,13 @@ public:
 
 	InvSumAggregator(const String& attr);
 
-	virtual void Apply(const Table::Ptr& table, const Value& row) override;
+	virtual void Apply(const Table::Ptr& table, const Value& row, AggregatorState **state) override;
-	virtual double GetResult(void) const override;
+	virtual double GetResultAndFreeState(AggregatorState *state) const override;
 
 private:
-	double m_InvSum;
 	String m_InvSumAttr;
+
+	static InvSumAggregatorState *EnsureState(AggregatorState **state);
 };
 
 }

lib/livestatus/livestatusquery.cpp

@@ -512,18 +512,35 @@ void LivestatusQuery::ExecuteGetHelper(const Stream::Ptr& stream)
 			AppendResultRow(result, row, first_row);
 		}
 	} else {
-		std::vector<double> stats(m_Aggregators.size(), 0);
+		std::map<std::vector<Value>, std::vector<AggregatorState *> > allStats;
-		int index = 0;
 
 		/* add aggregated stats */
-		for (const Aggregator::Ptr aggregator : m_Aggregators) {
 		for (const LivestatusRowValue& object : objects) {
-				aggregator->Apply(table, object.Row);
+			Column column = table->GetColumn(m_Columns[0]);
+
+			std::vector<Value> statsKey;
+
+			for (const String& columnName : m_Columns) {
+				Column column = table->GetColumn(columnName);
+				statsKey.push_back(column.ExtractValue(object.Row, object.GroupByType, object.GroupByObject));
 			}
 
-			stats[index] = aggregator->GetResult();
+			auto it = allStats.find(statsKey);
+
+			if (it == allStats.end()) {
+				std::vector<AggregatorState *> newStats(m_Aggregators.size(), NULL);
+				it = allStats.insert(std::make_pair(statsKey, newStats)).first;
+			}
+
+			auto& stats = it->second;
+
+			int index = 0;
+
+			for (const Aggregator::Ptr aggregator : m_Aggregators) {
+				aggregator->Apply(table, object.Row, &stats[index]);
 				index++;
 			}
+		}
 
 		/* add column headers both for raw and aggregated data */
 		if (m_ColumnHeaders) {
@@ -540,29 +557,23 @@ void LivestatusQuery::ExecuteGetHelper(const Stream::Ptr& stream)
 			AppendResultRow(result, header, first_row);
 		}
 
+		for (const auto& kv : allStats) {
 			Array::Ptr row = new Array();
 
 			row->Reserve(m_Columns.size() + m_Aggregators.size());
 
-		/*
+			for (const Value& keyPart : kv.first) {
-		 * add selected columns next to stats
+				row->Add(keyPart);
-		 * may not be accurate for grouping!
-		 */
-		if (objects.size() > 0 && m_Columns.size() > 0) {
-			for (const String& columnName : m_Columns) {
-				Column column = table->GetColumn(columnName);
-
-				LivestatusRowValue object = objects[0]; //first object wins
-
-				row->Add(column.ExtractValue(object.Row, object.GroupByType, object.GroupByObject));
-			}
 			}
 
+			auto& stats = kv.second;
+
 			for (size_t i = 0; i < m_Aggregators.size(); i++)
-			row->Add(stats[i]);
+				row->Add(m_Aggregators[i]->GetResultAndFreeState(stats[i]));
 
 			AppendResultRow(result, row, first_row);
 		}
+	}
 
 	EndResultSet(result);
 

lib/livestatus/maxaggregator.cpp

@@ -22,20 +22,34 @@
 using namespace icinga;
 
 MaxAggregator::MaxAggregator(const String& attr)
-    : m_Max(0), m_MaxAttr(attr)
+    : m_MaxAttr(attr)
 { }
 
-void MaxAggregator::Apply(const Table::Ptr& table, const Value& row)
+MaxAggregatorState *MaxAggregator::EnsureState(AggregatorState **state)
+{
+	if (!*state)
+		*state = new MaxAggregatorState();
+
+	return static_cast<MaxAggregatorState *>(*state);
+}
+
+void MaxAggregator::Apply(const Table::Ptr& table, const Value& row, AggregatorState **state)
 {
 	Column column = table->GetColumn(m_MaxAttr);
 
 	Value value = column.ExtractValue(row);
 
-	if (value > m_Max)
+	MaxAggregatorState *pstate = EnsureState(state);
-		m_Max = value;
+
+	if (value > pstate->Max)
+		pstate->Max = value;
 }
 
-double MaxAggregator::GetResult(void) const
+double MaxAggregator::GetResultAndFreeState(AggregatorState *state) const
 {
-	return m_Max;
+	MaxAggregatorState *pstate = EnsureState(&state);
+	double result = pstate->Max;
+	delete pstate;
+
+	return result;
 }

lib/livestatus/maxaggregator.hpp

@@ -26,6 +26,18 @@
 namespace icinga
 {
 
+/**
+ * @ingroup livestatus
+ */
+struct I2_LIVESTATUS_API MaxAggregatorState : public AggregatorState
+{
+	MaxAggregatorState(void)
+	    : Max(0)
+	{ }
+
+	double Max;
+};
+
 /**
  * @ingroup livestatus
  */
@@ -36,12 +48,13 @@ public:
 
 	MaxAggregator(const String& attr);
 
-	virtual void Apply(const Table::Ptr& table, const Value& row) override;
+	virtual void Apply(const Table::Ptr& table, const Value& row, AggregatorState **state) override;
-	virtual double GetResult(void) const override;
+	virtual double GetResultAndFreeState(AggregatorState *state) const override;
 
 private:
-	double m_Max;
 	String m_MaxAttr;
+
+	static MaxAggregatorState *EnsureState(AggregatorState **state);
 };
 
 }

lib/livestatus/minaggregator.cpp

@@ -22,23 +22,41 @@
 using namespace icinga;
 
 MinAggregator::MinAggregator(const String& attr)
-    : m_Min(DBL_MAX), m_MinAttr(attr)
+    : m_MinAttr(attr)
 { }
 
-void MinAggregator::Apply(const Table::Ptr& table, const Value& row)
+MinAggregatorState *MinAggregator::EnsureState(AggregatorState **state)
+{
+	if (!*state)
+		*state = new MinAggregatorState();
+
+	return static_cast<MinAggregatorState *>(*state);
+}
+
+void MinAggregator::Apply(const Table::Ptr& table, const Value& row, AggregatorState **state)
 {
 	Column column = table->GetColumn(m_MinAttr);
 
 	Value value = column.ExtractValue(row);
 
-	if (value < m_Min)
+	MinAggregatorState *pstate = EnsureState(state);
-		m_Min = value;
+
+	if (value < pstate->Min)
+		pstate->Min = value;
 }
 
-double MinAggregator::GetResult(void) const
+double MinAggregator::GetResultAndFreeState(AggregatorState *state) const
 {
-	if (m_Min == DBL_MAX)
+	MinAggregatorState *pstate = EnsureState(&state);
-		return 0;
+
+	double result;
+
+	if (pstate->Min == DBL_MAX)
+		result = 0;
 	else
-		return m_Min;
+		result = pstate->Min;
+
+	delete pstate;
+
+	return result;
 }

lib/livestatus/minaggregator.hpp

@@ -27,6 +27,18 @@
 namespace icinga
 {
 
+/**
+ * @ingroup livestatus
+ */
+struct I2_LIVESTATUS_API MinAggregatorState : public AggregatorState
+{
+	MinAggregatorState(void)
+	    : Min(DBL_MAX)
+	{ }
+
+	double Min;
+};
+
 /**
  * @ingroup livestatus
  */
@@ -37,12 +49,13 @@ public:
 
 	MinAggregator(const String& attr);
 
-	virtual void Apply(const Table::Ptr& table, const Value& row) override;
+	virtual void Apply(const Table::Ptr& table, const Value& row, AggregatorState **state) override;
-	virtual double GetResult(void) const override;
+	virtual double GetResultAndFreeState(AggregatorState *state) const override;
 
 private:
-	double m_Min;
 	String m_MinAttr;
+
+	static MinAggregatorState *EnsureState(AggregatorState **state);
 };
 
 }

lib/livestatus/stdaggregator.cpp

@@ -23,21 +23,35 @@
 using namespace icinga;
 
 StdAggregator::StdAggregator(const String& attr)
-    : m_StdSum(0), m_StdQSum(0), m_StdCount(0), m_StdAttr(attr)
+    : m_StdAttr(attr)
 { }
 
-void StdAggregator::Apply(const Table::Ptr& table, const Value& row)
+StdAggregatorState *StdAggregator::EnsureState(AggregatorState **state)
+{
+	if (!*state)
+		*state = new StdAggregatorState();
+
+	return static_cast<StdAggregatorState *>(*state);
+}
+
+void StdAggregator::Apply(const Table::Ptr& table, const Value& row, AggregatorState **state)
 {
 	Column column = table->GetColumn(m_StdAttr);
 
 	Value value = column.ExtractValue(row);
 
-	m_StdSum += value;
+	StdAggregatorState *pstate = EnsureState(state);
-	m_StdQSum += pow(value, 2);
+
-	m_StdCount++;
+	pstate->StdSum += value;
+	pstate->StdQSum += pow(value, 2);
+	pstate->StdCount++;
 }
 
-double StdAggregator::GetResult(void) const
+double StdAggregator::GetResultAndFreeState(AggregatorState *state) const
 {
-	return sqrt((m_StdQSum - (1 / m_StdCount) * pow(m_StdSum, 2)) / (m_StdCount - 1));
+	StdAggregatorState *pstate = EnsureState(&state);
+	double result = sqrt((pstate->StdQSum - (1 / pstate->StdCount) * pow(pstate->StdSum, 2)) / (pstate->StdCount - 1));
+	delete pstate;
+
+	return result;
 }

lib/livestatus/stdaggregator.hpp

@@ -26,6 +26,20 @@
 namespace icinga
 {
 
+/**
+ * @ingroup livestatus
+ */
+struct I2_LIVESTATUS_API StdAggregatorState : public AggregatorState
+{
+	StdAggregatorState(void)
+	    : StdSum(0), StdQSum(0), StdCount(0)
+	{ }
+
+	double StdSum;
+	double StdQSum;
+	double StdCount;
+};
+
 /**
  * @ingroup livestatus
  */
@@ -36,14 +50,13 @@ public:
 
 	StdAggregator(const String& attr);
 
-	virtual void Apply(const Table::Ptr& table, const Value& row) override;
+	virtual void Apply(const Table::Ptr& table, const Value& row, AggregatorState **state) override;
-	virtual double GetResult(void) const override;
+	virtual double GetResultAndFreeState(AggregatorState *state) const override;
 
 private:
-	double m_StdSum;
-	double m_StdQSum;
-	double m_StdCount;
 	String m_StdAttr;
+
+	static StdAggregatorState *EnsureState(AggregatorState **state);
 };
 
 }

lib/livestatus/sumaggregator.cpp

@@ -22,19 +22,33 @@
 using namespace icinga;
 
 SumAggregator::SumAggregator(const String& attr)
-    : m_Sum(0), m_SumAttr(attr)
+    : m_SumAttr(attr)
 { }
 
-void SumAggregator::Apply(const Table::Ptr& table, const Value& row)
+SumAggregatorState *SumAggregator::EnsureState(AggregatorState **state)
+{
+	if (!*state)
+		*state = new SumAggregatorState();
+
+	return static_cast<SumAggregatorState *>(*state);
+}
+
+void SumAggregator::Apply(const Table::Ptr& table, const Value& row, AggregatorState **state)
 {
 	Column column = table->GetColumn(m_SumAttr);
 
 	Value value = column.ExtractValue(row);
 
-	m_Sum += value;
+	SumAggregatorState *pstate = EnsureState(state);
+
+	pstate->Sum += value;
 }
 
-double SumAggregator::GetResult(void) const
+double SumAggregator::GetResultAndFreeState(AggregatorState *state) const
 {
-	return m_Sum;
+	SumAggregatorState *pstate = EnsureState(&state);
+	double result = pstate->Sum;
+	delete pstate;
+
+	return result;
 }

lib/livestatus/sumaggregator.hpp

@@ -26,6 +26,18 @@
 namespace icinga
 {
 
+/**
+ * @ingroup livestatus
+ */
+struct I2_LIVESTATUS_API SumAggregatorState : public AggregatorState
+{
+	SumAggregatorState(void)
+	    : Sum(0)
+	{ }
+
+	double Sum;
+};
+
 /**
  * @ingroup livestatus
  */
@@ -36,12 +48,13 @@ public:
 
 	SumAggregator(const String& attr);
 
-	virtual void Apply(const Table::Ptr& table, const Value& row) override;
+	virtual void Apply(const Table::Ptr& table, const Value& row, AggregatorState **state) override;
-	virtual double GetResult(void) const override;
+	virtual double GetResultAndFreeState(AggregatorState *state) const override;
 
 private:
-	double m_Sum;
 	String m_SumAttr;
+
+	static SumAggregatorState *EnsureState(AggregatorState **state);
 };
 
 }