Add debug logs to ingest (#246)

* added debug output of decoded records in ingest

* fixed import

Author: KalmanMeth

docs/api.md

@@ -63,6 +63,8 @@ Following is the supported API format for the kafka ingest:
              type: (enum) one of the following:
                  json: JSON decoder
                  protobuf: Protobuf decoder
+         batchMaxLen: the number of accumulated flows before being forwarded for processing
+         commitInterval: the interval (in milliseconds) at which offsets are committed to the broker.  If 0, commits will be handled synchronously.
 </pre>
 ## Ingest GRPC from Network Observability eBPF Agent
 Following is the supported API format for the Network Observability eBPF ingest:

pkg/pipeline/ingest/ingest_collector.go

@@ -174,6 +174,7 @@ func (ingestC *ingestCollector) processLogLines(out chan<- []config.GenericMap)
 				log.Debugf("ingestCollector sending %d entries, %d entries waiting", len(records), len(ingestC.in))
 				linesProcessed.Add(float64(len(records)))
 				queueLength.Set(float64(len(out)))
+				log.Debugf("ingestCollector records = %v", records)
 				out <- records
 				records = []config.GenericMap{}
 			}
@@ -189,6 +190,7 @@ func (ingestC *ingestCollector) processLogLines(out chan<- []config.GenericMap)
 				log.Debugf("ingestCollector sending %d entries, %d entries waiting", len(records), len(ingestC.in))
 				linesProcessed.Add(float64(len(records)))
 				queueLength.Set(float64(len(out)))
+				log.Debugf("ingestCollector records = %v", records)
 				out <- records
 				records = []config.GenericMap{}
 			}

pkg/pipeline/ingest/ingest_file.go

@@ -64,6 +64,7 @@ func (ingestF *IngestFile) Ingest(out chan<- []config.GenericMap) {
 		lines = append(lines, text)
 	}
 	decoded := ingestF.decoder.Decode(lines)
+	log.Debugf("IngestFile decoded = %v", decoded)
 
 	log.Debugf("Ingesting %d log lines from %s", len(lines), filename)
 	switch ingestF.params.Type {

pkg/pipeline/ingest/ingest_grpc.go

@@ -15,6 +15,7 @@ import (
 	"github.com/netobserv/netobserv-ebpf-agent/pkg/pbflow"
 	flow "github.com/netsampler/goflow2/utils"
 	"github.com/prometheus/client_golang/prometheus"
+	log "github.com/sirupsen/logrus"
 	grpc2 "google.golang.org/grpc"
 	"google.golang.org/grpc/metadata"
 	"google.golang.org/protobuf/proto"
@@ -79,7 +80,9 @@ func (no *GRPCProtobuf) Ingest(out chan<- []config.GenericMap) {
 		no.collector.Close()
 	}()
 	for fp := range no.flowPackets {
-		out <- no.decoder.Decode([]interface{}{fp})
+		records := no.decoder.Decode([]interface{}{fp})
+		log.Debugf("GRPCProtobuf records = %v", records)
+		out <- records
 	}
 }
 

pkg/pipeline/transform/transform_generic.go

@@ -33,6 +33,7 @@ func (g *Generic) Transform(input []config.GenericMap) []config.GenericMap {
 	log.Debugf("entering Generic Transform g = %v", g)
 	output := make([]config.GenericMap, 0)
 	for _, entry := range input {
+		log.Debugf("entry.GenericMap = %v", entry)
 		outputEntry := make(config.GenericMap)
 		if g.policy != "replace_keys" {
 			// copy old map to new map