Fix invalid ingress counter for dnstap collector (#839)

* fix: invalid ingress counter for dnstap collector

* Update README.md

README.md

@@ -1,7 +1,7 @@
 <p align="center">
 <img src="https://goreportcard.com/badge/github.com/dmachard/go-dns-collector" alt="Go Report"/>
 <img src="https://img.shields.io/badge/go%20version-min%201.21-green" alt="Go version"/>
-<img src="https://img.shields.io/badge/go%20tests-506-green" alt="Go tests"/>
+<img src="https://img.shields.io/badge/go%20tests-507-green" alt="Go tests"/>
 <img src="https://img.shields.io/badge/go%20bench-21-green" alt="Go bench"/>
 <img src="https://img.shields.io/badge/go%20lines-31679-green" alt="Go lines"/>
 </p>

workers/dnstapserver.go

@@ -344,9 +344,6 @@ func (w *DNSTapProcessor) StartCollect() {
 				continue
 			}
 
-			// count global messages
-			w.CountIngressTraffic()
-
 			// init dns message
 			dm := dnsutils.DNSMessage{}
 			dm.Init()

workers/dnstapserver_test.go

@@ -11,6 +11,7 @@ import (
 
 	"github.com/dmachard/go-dnscollector/dnsutils"
 	"github.com/dmachard/go-dnscollector/pkgconfig"
+	"github.com/dmachard/go-dnscollector/telemetry"
 	"github.com/dmachard/go-dnstap-protobuf"
 	"github.com/dmachard/go-framestream"
 	"github.com/dmachard/go-logger"
@@ -240,7 +241,7 @@ func Test_DnstapProcessor_toDNSMessage(t *testing.T) {
 	}
 }
 
-func Test_DnstapProcessor_DecodeCounters(t *testing.T) {
+func Test_DnstapProcessor_DecodeDNSCounters(t *testing.T) {
 	logger := logger.New(true)
 	var o bytes.Buffer
 	logger.SetOutput(&o)
@@ -622,3 +623,54 @@ func Test_DnstapProcessor_BufferLoggerIsFull(t *testing.T) {
 		t.Errorf("invalid qname in second dns message: %s", dm2.DNS.Qname)
 	}
 }
+
+// test for telemetry counter
+func Test_DnstapProcessor_TelemetryCounters(t *testing.T) {
+	logger := logger.New(true)
+	var o bytes.Buffer
+	logger.SetOutput(&o)
+
+	// run the consumer with a fake logger
+	fl := GetWorkerForTest(pkgconfig.DefaultBufferSize)
+
+	cfg := pkgconfig.GetDefaultConfig()
+	cfg.Global.Telemetry.Enabled = true
+	cfg.Global.Worker.InternalMonitor = 1
+
+	config := pkgconfig.Config{}
+	metrics := telemetry.NewPrometheusCollector(&config)
+
+	// init the dnstap consumer
+	consumer := NewDNSTapProcessor(0, "peertest", cfg, logger, "test", 512)
+	consumer.SetMetrics(metrics)
+	consumer.AddDefaultRoute(fl)
+	consumer.AddDroppedRoute(fl)
+
+	// get dns packet
+	queryPkt, _ := dnsutils.GetFakeDNS()
+
+	// prepare dnstap
+	dt := &dnstap.Dnstap{}
+	dt.Type = dnstap.Dnstap_Type.Enum(1)
+
+	dt.Message = &dnstap.Message{}
+	dt.Message.Type = dnstap.Message_Type.Enum(5) // CLIENT_QUERY
+	dt.Message.ResponseMessage = queryPkt
+	data, _ := proto.Marshal(dt)
+
+	// start the consumer
+	go consumer.StartCollect()
+
+	// add packet to consumer and read output
+	consumer.GetDataChannel() <- data
+
+	<-fl.GetInputChannel()
+	r := <-metrics.Record
+
+	if r.TotalIngress != 1 {
+		t.Errorf("invalid total ingress oucnter: got %d expect 1", r.TotalIngress)
+	}
+	if r.TotalEgress != 1 {
+		t.Errorf("invalid total egress counter: got %d expect 1", r.TotalEgress)
+	}
+}

workers/worker.go

@@ -232,7 +232,7 @@ func (w *GenericWorker) Monitor() {
 
 			// // send to telemetry?
 			if w.config.Global.Telemetry.Enabled && w.metrics != nil {
-				if w.totalIngress > 0 || w.totalForwarded > 0 || w.totalDropped > 0 {
+				if w.totalIngress > 0 || w.totalEgress > 0 || w.totalForwarded > 0 || w.totalDropped > 0 {
 					w.metrics.Record <- telemetry.WorkerStats{
 						Name:                 w.GetName(),
 						TotalIngress:         w.totalIngress,