review

processor/processor.go

@@ -1125,6 +1125,30 @@ func (proc *Handle) getNonSuccessfulMetrics(
 	return m
 }
 
+func procFilteredCountStat(destType, pu, statusCode string) {
+	stats.Default.NewTaggedStat(
+		"proc_filtered_counts",
+		stats.CountType,
+		stats.Tags{
+			"destName":   destType,
+			"statusCode": statusCode,
+			"stage":      pu,
+		},
+	).Increment()
+}
+
+func procErrorCountsStat(destType, pu, statusCode string) {
+	stats.Default.NewTaggedStat(
+		"proc_error_counts",
+		stats.CountType,
+		stats.Tags{
+			"destName":   destType,
+			"statusCode": statusCode,
+			"stage":      pu,
+		},
+	).Increment()
+}
+
 func (proc *Handle) getTransformationMetrics(
 	transformerResponses []transformer.TransformerResponse,
 	state string,
@@ -1137,9 +1161,9 @@ func (proc *Handle) getTransformationMetrics(
 	statusDetailsMap := make(map[string]map[string]*types.StatusDetail)
 	countMap := make(map[string]int64)
 	var jobs []*jobsdb.JobT
-	isDropped := "false"
+	statFunc := procErrorCountsStat
 	if state == jobsdb.Filtered.State {
-		isDropped = "true"
+		statFunc = procFilteredCountStat
 	}
 	for i := range transformerResponses {
 		failedEvent := &transformerResponses[i]
@@ -1215,14 +1239,7 @@ func (proc *Handle) getTransformationMetrics(
 		}
 		jobs = append(jobs, &newFailedJob)
 
-		procErrorStat := stats.Default.NewTaggedStat("proc_error_counts", stats.CountType, stats.Tags{
-			"destName":   commonMetaData.DestinationType,
-			"statusCode": strconv.Itoa(failedEvent.StatusCode),
-			"stage":      pu,
-			"isDropped":  isDropped,
-		})
-
-		procErrorStat.Increment()
+		statFunc(commonMetaData.DestinationType, pu, strconv.Itoa(failedEvent.StatusCode))
 	}
 
 	// REPORTING - START