Add logging

go/cron/kafka_submitter.go

@@ -7,6 +7,7 @@ import (
 	"sync"
 	"time"
 
+	"github.com/inverse-inc/go-utils/log"
 	"github.com/segmentio/kafka-go"
 	"github.com/segmentio/kafka-go/compress"
 	"github.com/segmentio/kafka-go/sasl/plain"
@@ -117,13 +118,19 @@ func (s *KafkaSubmiter) send(events []*NetworkEvent) {
 		filteredEvents = events
 	}
 
+	ctx := context.Background()
 	messages := make([]kafka.Message, 0, len(filteredEvents))
 	for i := 0; i < len(filteredEvents); i++ {
 		data, err := json.Marshal(events[i])
 		if err != nil {
 			//TODO log error
 			continue
 		}
+		log.LogInfof(
+			ctx,
+			"Sending SrcIp: %s, DstIp: %s, DstPort: %d, Direction: %s",
+			events[i].SourceIp, events[i].DestIp, events[i].DestPort, events[i].Direction,
+		)
 		messages = append(messages, kafka.Message{Value: data})
 	}