From 1a9fbd2713a034a42caa018764a136970e5f6251 Mon Sep 17 00:00:00 2001
From: Vishal Raj <vishal.raj@elastic.co>
Date: Fri, 28 Jul 2023 15:14:27 +0800
Subject: [PATCH] Check length of binary estimator to find overflow (#46)

---
 aggregators/converter.go | 22 +++++++++++-----------
 aggregators/merger.go    |  2 +-
 2 files changed, 12 insertions(+), 12 deletions(-)

diff --git a/aggregators/converter.go b/aggregators/converter.go
index b485122..c3f85c7 100644
--- a/aggregators/converter.go
+++ b/aggregators/converter.go
@@ -138,7 +138,7 @@ func CombinedMetricsToBatch(
 
 	batchSize := 0
 	// service_summary overflow metric
-	if cm.OverflowServiceInstancesEstimator != nil {
+	if len(cm.OverflowServiceInstancesEstimator) > 0 {
 		batchSize++
 	}
 
@@ -156,13 +156,13 @@ func CombinedMetricsToBatch(
 		if sm.OverflowGroups == nil {
 			continue
 		}
-		if sm.OverflowGroups.OverflowTransactionsEstimator != nil {
+		if len(sm.OverflowGroups.OverflowTransactionsEstimator) > 0 {
 			batchSize++
 		}
-		if sm.OverflowGroups.OverflowServiceTransactionsEstimator != nil {
+		if len(sm.OverflowGroups.OverflowServiceTransactionsEstimator) > 0 {
 			batchSize++
 		}
-		if sm.OverflowGroups.OverflowSpansEstimator != nil {
+		if len(sm.OverflowGroups.OverflowSpansEstimator) > 0 {
 			batchSize++
 		}
 	}
@@ -213,7 +213,7 @@ func CombinedMetricsToBatch(
 		if sm.OverflowGroups == nil {
 			continue
 		}
-		if sm.OverflowGroups.OverflowTransactionsEstimator != nil {
+		if len(sm.OverflowGroups.OverflowTransactionsEstimator) > 0 {
 			estimator := hllSketch(sm.OverflowGroups.OverflowTransactionsEstimator)
 			event := getBaseEvent(sk)
 			overflowTxnMetricsToAPMEvent(
@@ -225,7 +225,7 @@ func CombinedMetricsToBatch(
 			)
 			b = append(b, event)
 		}
-		if sm.OverflowGroups.OverflowServiceTransactionsEstimator != nil {
+		if len(sm.OverflowGroups.OverflowServiceTransactionsEstimator) > 0 {
 			estimator := hllSketch(
 				sm.OverflowGroups.OverflowServiceTransactionsEstimator,
 			)
@@ -239,7 +239,7 @@ func CombinedMetricsToBatch(
 			)
 			b = append(b, event)
 		}
-		if sm.OverflowGroups.OverflowSpansEstimator != nil {
+		if len(sm.OverflowGroups.OverflowSpansEstimator) > 0 {
 			estimator := hllSketch(sm.OverflowGroups.OverflowSpansEstimator)
 			event := getBaseEvent(sk)
 			overflowSpanMetricsToAPMEvent(
@@ -252,7 +252,7 @@ func CombinedMetricsToBatch(
 			b = append(b, event)
 		}
 	}
-	if cm.OverflowServiceInstancesEstimator != nil {
+	if len(cm.OverflowServiceInstancesEstimator) > 0 {
 		estimator := hllSketch(cm.OverflowServiceInstancesEstimator)
 		getOverflowBaseEvent := func() *modelpb.APMEvent {
 			return &modelpb.APMEvent{
@@ -270,7 +270,7 @@ func CombinedMetricsToBatch(
 			aggIntervalStr,
 		)
 		b = append(b, event)
-		if cm.OverflowServices.OverflowTransactionsEstimator != nil {
+		if len(cm.OverflowServices.OverflowTransactionsEstimator) > 0 {
 			estimator := hllSketch(cm.OverflowServices.OverflowTransactionsEstimator)
 			event := getOverflowBaseEvent()
 			overflowTxnMetricsToAPMEvent(
@@ -283,7 +283,7 @@ func CombinedMetricsToBatch(
 			b = append(b, event)
 
 		}
-		if cm.OverflowServices.OverflowServiceTransactionsEstimator != nil {
+		if len(cm.OverflowServices.OverflowServiceTransactionsEstimator) > 0 {
 			estimator := hllSketch(
 				cm.OverflowServices.OverflowServiceTransactionsEstimator,
 			)
@@ -297,7 +297,7 @@ func CombinedMetricsToBatch(
 			)
 			b = append(b, event)
 		}
-		if cm.OverflowServices.OverflowSpansEstimator != nil {
+		if len(cm.OverflowServices.OverflowSpansEstimator) > 0 {
 			estimator := hllSketch(cm.OverflowServices.OverflowSpansEstimator)
 			event := getOverflowBaseEvent()
 			overflowSpanMetricsToAPMEvent(
diff --git a/aggregators/merger.go b/aggregators/merger.go
index aaea3c0..e4792ca 100644
--- a/aggregators/merger.go
+++ b/aggregators/merger.go
@@ -54,7 +54,7 @@ func (m *combinedMetricsMerger) merge(from *aggregationpb.CombinedMetrics) {
 	}
 	// If there is overflow due to max services in either of the buckets being
 	// merged then we can merge the overflow buckets without considering any other scenarios.
-	if from.OverflowServiceInstancesEstimator != nil {
+	if len(from.OverflowServiceInstancesEstimator) > 0 {
 		mergeOverflow(&m.metrics.OverflowServices, from.OverflowServices)
 		mergeEstimator(
 			&m.metrics.OverflowServiceInstancesEstimator,