chore: send multiple reports in a single request to reporting

rudderlabs · Nov 6, 2024 · 19e9cfb · 19e9cfb
1 parent 9f2b7e7
commit 19e9cfb
Showing 1 changed file with 8 additions and 4 deletions.
diff --git a/enterprise/reporting/reporting.go b/enterprise/reporting/reporting.go
@@ -74,6 +74,7 @@ type DefaultReporter struct {
 	getReportsQueryTime       stats.Measurement
 	requestLatency            stats.Measurement
 	stats                     stats.Stats
+	maxReportsCountInARequest config.ValueLoader[int]
 }
 
 func NewDefaultReporter(ctx context.Context, log logger.Logger, configSubscriber *configSubscriber, stats stats.Stats) *DefaultReporter {
@@ -88,6 +89,7 @@ func NewDefaultReporter(ctx context.Context, log logger.Logger, configSubscriber
 	maxConcurrentRequests := config.GetReloadableIntVar(32, 1, "Reporting.maxConcurrentRequests")
 	maxOpenConnections := config.GetIntVar(32, 1, "Reporting.maxOpenConnections")
 	dbQueryTimeout = config.GetReloadableDurationVar(60, time.Second, "Reporting.dbQueryTimeout")
+	maxReportsCountInARequest := config.GetReloadableIntVar(1, 1, "Reporting.maxReportsCountInARequest")
 	// only send reports for wh actions sources if whActionsOnly is configured
 	whActionsOnly := config.GetBool("REPORTING_WH_ACTIONS_ONLY", false)
 	if whActionsOnly {
@@ -114,6 +116,7 @@ func NewDefaultReporter(ctx context.Context, log logger.Logger, configSubscriber
 		maxOpenConnections:                   maxOpenConnections,
 		maxConcurrentRequests:                maxConcurrentRequests,
 		dbQueryTimeout:                       dbQueryTimeout,
+		maxReportsCountInARequest:            maxReportsCountInARequest,
 		stats:                                stats,
 	}
 }
@@ -265,7 +268,7 @@ func (r *DefaultReporter) getReports(currentMs int64, syncerKey string) (reports
 	return metricReports, queryMin.Int64, err
 }
 
-func (*DefaultReporter) getAggregatedReports(reports []*types.ReportByStatus) []*types.Metric {
+func (r *DefaultReporter) getAggregatedReports(reports []*types.ReportByStatus) []*types.Metric {
 	metricsByGroup := map[string]*types.Metric{}
 	var values []*types.Metric
 
@@ -282,9 +285,6 @@ func (*DefaultReporter) getAggregatedReports(reports []*types.ReportByStatus) []
 			report.ConnectionDetails.TrackingPlanID,
 			strconv.Itoa(report.ConnectionDetails.TrackingPlanVersion),
 			report.PUDetails.InPU, report.PUDetails.PU,
-			report.StatusDetail.Status,
-			strconv.Itoa(report.StatusDetail.StatusCode),
-			report.StatusDetail.EventName, report.StatusDetail.EventType,
 		}
 		return strings.Join(groupingIdentifiers, `::`)
 	}
@@ -336,6 +336,10 @@ func (*DefaultReporter) getAggregatedReports(reports []*types.ReportByStatus) []
 			EventType:      report.StatusDetail.EventType,
 			ErrorType:      report.StatusDetail.ErrorType,
 		})
+
+		if len(metricsByGroup[identifier].StatusDetails) >= r.maxReportsCountInARequest.Load() {
+			delete(metricsByGroup, identifier)
+		}
 	}
 
 	return values