Merge pull request #250 from OlivierCazade/kafka-metrics

Added metrics to kafka ingest
netobserv · Jul 6, 2022 · 467e409 · 467e409
2 parents b350404 + e277183
commit 467e409
Showing 1 changed file with 5 additions and 1 deletion.
diff --git a/pkg/pipeline/ingest/ingest_kafka.go b/pkg/pipeline/ingest/ingest_kafka.go
@@ -97,10 +97,12 @@ func (ingestK *ingestKafka) processLogLines(out chan<- []config.GenericMap) {
 			if len(records) >= ingestK.batchMaxLength {
 				log.Debugf("ingestKafka sending %d records, %d entries waiting", len(records), len(ingestK.in))
 				decoded := ingestK.decoder.Decode(records)
-				out <- decoded
+				linesProcessed.Add(float64(len(records)))
+				queueLength.Set(float64(len(out)))
 				ingestK.prevRecords = decoded
 				log.Debugf("prevRecords = %v", ingestK.prevRecords)
 				records = []interface{}{}
+				out <- decoded
 			}
 		case <-flushRecords.C: // Maximum batch time for each batch
 			// Process batch of records (if not empty)
@@ -113,6 +115,8 @@ func (ingestK *ingestKafka) processLogLines(out chan<- []config.GenericMap) {
 				}
 				log.Debugf("ingestKafka sending %d records, %d entries waiting", len(records), len(ingestK.in))
 				decoded := ingestK.decoder.Decode(records)
+				linesProcessed.Add(float64(len(records)))
+				queueLength.Set(float64(len(out)))
 				ingestK.prevRecords = decoded
 				log.Debugf("prevRecords = %v", ingestK.prevRecords)
 				out <- decoded