dmachard · dmachard · Mar 6, 2024 · Mar 6, 2024 · Mar 6, 2024 · Mar 6, 2024
diff --git a/.github/workflows/bench-go.yml b/.github/workflows/bench-go.yml
@@ -48,7 +48,12 @@ jobs:
         go version
         sudo go version
 
-    - name: Bench
+    - name: Bench dnsutils
       run: |
         cd dnsutils/
-        go test -run=^$ -bench=.
+        go test -benchmem -run=^$ -bench=.
+
+    - name: Bench transformers
+      run: |
+        cd transformers/
+        go test -benchmem -run=^$ -bench=^BenchmarkUserPrivacy.*\|BenchmarkTransforms.*$
diff --git a/.gitignore b/.gitignore
@@ -22,4 +22,5 @@ go-dnscollector
 bin/
 include/
 docs/_integration/elasticsearch/data/
-docs/_integration/kafka/data/
+docs/_integration/kafka/data/
+docs/_integration/fluentd/data/
diff --git a/README.md b/README.md
@@ -2,7 +2,7 @@
 <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.20-green" alt="Go version"/>
 <img src="https://img.shields.io/badge/go%20tests-429-green" alt="Go tests"/>
-<img src="https://img.shields.io/badge/go%20tests-8-green" alt="Go bench"/>
+<img src="https://img.shields.io/badge/go%20bench-14-green" alt="Go bench"/>
 <img src="https://img.shields.io/badge/go%20lines-38661-green" alt="Go lines"/>
 </p>
 
@@ -111,6 +111,7 @@ The [`_examples`](./docs/_examples) folder from documentation contains a number
 
 The [`_integration`](./docs/_integration) folder contains DNS-collector `configuration` files and `docker compose` examples for popular tools:
 
+- [Fluentd](./docs/_integration/fluentd/README.md)
 - [Elasticsearch](./docs/_integration/elasticsearch/README.md)
 - [Kafka](./docs/_integration/kafka/README.md)
 

diff --git a/config.yml b/config.yml
@@ -595,7 +595,7 @@ multiplexer:
 #   # how many DNS messages will be buffered before being sent
 #   buffer-size: 100
 #   # Channel buffer size for incoming packets, number of packet before to drop it.
-#   chan-buffer-size: 65535
+#   chan-buffer-size: 4096
 
 # # resend captured dns traffic to a InfluxDB database
 # influxdb:

diff --git a/dnsutils/message_test.go b/dnsutils/message_test.go
@@ -13,6 +13,16 @@ import (
 	"google.golang.org/protobuf/proto"
 )
 
+// Bench to init DNS message
+func BenchmarkDnsMessage_Init(b *testing.B) {
+	b.ResetTimer()
+	for i := 0; i < b.N; i++ {
+		dm := DNSMessage{}
+		dm.Init()
+		dm.InitTransforms()
+	}
+}
+
 // Tests for DNSTap format
 func encodeToDNSTap(dm DNSMessage, t *testing.T) *ExtendedDnstap {
 	// encode to extended dnstap

diff --git a/docs/_integration/elasticsearch/README.md b/docs/_integration/elasticsearch/README.md
@@ -1,7 +1,11 @@
 
 # DNS-collector with Elastic and Kibana
 
-- Copy folder [./docs/_integration/elasticsearch] and start the docker stack:
+- Copy this [folder](./docs/_integration/elasticsearch).
+
+- Create the `data` folder.
+
+- Start the docker stack:
 
     ```bash
     sudo docker compose up -d
@@ -13,7 +17,7 @@
 
 - Finally create index pattern `dnscollector` and choose `dnstap.timestamp-rfc33939ns`
 
-- Run DNScollector from source:
+- Finally, run DNScollector from source and generate some DNS logs from your DNS server with DNStap protocol.
 
     ```bash
     go run . -config docs/_integration/elasticsearch/config.yml

diff --git a/docs/_integration/fluentd/README.md b/docs/_integration/fluentd/README.md
@@ -0,0 +1,26 @@
+
+# DNS-collector with Fluentd
+
+- Copy this [folder](./docs/_integration/fluentd)
+
+- Create the `data` folder.
+
+- Start the docker stack:
+
+    ```bash
+    sudo docker compose up -d
+
+    sudo docker compose logs
+    ...
+    fluentd  | 2024-03-06 05:46:12.930048059 +0000 fluent.info: {"port":24224,"bind":"0.0.0.0","message":"[input1] listening port port=24224 bind=\"0.0.0.0\""}
+    fluentd  | 2024-03-06 05:46:12 +0000 [warn]: #0 no patterns matched tag="fluent.info"
+    fluentd  | 2024-03-06 05:46:12.933055666 +0000 fluent.info: {"worker":0,"message":"fluentd worker is now running worker=0"}
+    ```
+
+- Finally, run DNScollector from source and generate some DNS logs from your DNS server with DNStap protocol.
+
+    ```bash
+    go run . -config docs/_integration/fluentd/config.yml
+    ```
+
+- Logs are available in ./data
diff --git a/docs/_integration/fluentd/config.yml b/docs/_integration/fluentd/config.yml
@@ -0,0 +1,33 @@
+
+global:
+  trace:
+    verbose: true
+
+multiplexer:
+  collectors:
+    - name: tap
+      dnstap:
+        listen-ip: 0.0.0.0
+        listen-port: 6000
+        chan-buffer-size: 4096
+  loggers:
+    - name: fluentd
+      fluentd:
+        transport: tcp
+        remote-address: 127.0.0.1
+        remote-port: 24224
+        connect-timeout: 5
+        retry-interval: 10
+        flush-interval: 30
+        tag: "dns.collector"
+        tls-insecure: false
+        tls-min-version: 1.2
+        ca-file: ""
+        cert-file: ""
+        key-file: ""
+        buffer-size: 100
+        chan-buffer-size: 4096
+
+  routes:
+    - from: [ tap ]
+      to: [ fluentd ]
diff --git a/docs/_integration/fluentd/docker-compose.yml b/docs/_integration/fluentd/docker-compose.yml
@@ -0,0 +1,13 @@
+version: "3.8"
+
+services:
+  fluentd:
+    container_name: fluentd
+    image: fluent/fluentd:v1.16-debian-2
+    user: 1000:1000
+    volumes:
+      - ./data:/fluentd/log
+    ports:
+      - "24224:24224/tcp"
+      - "24224:24224/udp"
+    restart: unless-stopped
diff --git a/docs/_integration/kafka/README.md b/docs/_integration/kafka/README.md
@@ -1,17 +1,21 @@
 
 # DNS-collector with Kafka
 
-- Copy folder [./docs/_integration/kafka] and start the docker stack:
+- Copy this [folder](./docs/_integration/kafka).
+
+- Create the `data` folder.
+
+- Start the docker stack:
 
     ```bash
     sudo docker compose up -d
     ```
 
 - Go to Apache Kafka interface through `http://127.0.0.1:8080`
 
-- The `dnscollector` topics should be available.
+- The `dnscollector` topic should be available.
 
-- Finally, run DNScollector from source:
+- Finally, run DNScollector from source and generate some DNS logs from your DNS server with DNStap protocol.
 
     ```bash
     go run . -config docs/_integration/kafka/config.yml

diff --git a/docs/loggers/logger_fluentd.md b/docs/loggers/logger_fluentd.md
@@ -2,46 +2,34 @@
 # Logger: Fluentd Client
 
 Fluentd client to remote server or unix socket.
-
-* to remote fluentd collector or unix socket
-* [msgpask](https://msgpack.org/)
-* tls support
+Based on [IBM/fluent-forward-go](https://github.com/IBM/fluent-forward-go) library
 
 Options:
 
-* `transport`: (string) network transport to use: `tcp`|`unix`|`tcp+tls`
-* `remote-address`: (string) remote address
-* `remote-port`: (integer) remote tcp port
-* `sock-path` **DEPRECATED, replaced by remote-address**: (string) unix socket path
-* `connect-timeout`: (integer) connect timeout in second
-* `retry-interval`: (integer) interval in second between retry reconnect
-* `flush-interval`: (integer) interval in second before to flush the buffer
-* `tag`: (string) tag name
-* `tls-support` **DEPRECATED, replaced with tcp+tls flag on transport**: (boolean) enable tls
-* `tls-insecure`: (boolean) insecure skip verify
-* `tls-min-version`: (string) min tls version, default to 1.2
-* `ca-file`: (string) provide CA file to verify the server certificate
-* `cert-file`: (string) provide client certificate file for mTLS
-* `key-file`: (string) provide client private key file for mTLS
-* `buffer-size`: (integer) how many DNS messages will be buffered before being sent
-* `chan-buffer-size`: (integer) channel buffer size used on incoming dns message, number of messages before to drop it.
-
-Default values:
+- `transport` (string) network transport to use: `tcp`|`unix`|`tcp+tls`. Default to `tcp`.
+  > Specifies the transport ot use.
+- `remote-address` (string) remote address.
+  > Specifies the remote address to connect to. Default to `127.0.0.1`.
+- `remote-port` (integer) remote tcp port. Default to `24224`.
+  > Specifies the remote TCP port to connect to.
+- `connect-timeout` (integer) connect timeout in second. Default to `5` seconds.
+  > Specifies the maximum time to wait for a connection attempt to complete.
+- `retry-interval` (integer) interval in second between retry reconnect. Default to `10` seconds.
+  > Specifies the interval between attempts to reconnect in case of connection failure.
+- `flush-interval` (integer) interval in second before to flush the buffer. Default to `30` seconds.
+  > Specifies the interval between buffer flushes.
+- `tag` (string) tag name. Default to `dns.collector`.
+  > Specifies the tag to use.
+- `tls-insecure` (boolean) insecure skip verify. Default to `false`.
+  > If set to true, skip verification of server certificate.
+- `tls-min-version` (string) min tls version. Default to `1.2`.
+  > Specifies the minimum TLS version that the server will support.
+- `ca-file` (string) provide CA file to verify the server certificate. Default to `(empty)`.
+  > Specifies the path to the CA (Certificate Authority) file used to verify the server's certificate.
+- `cert-file` (string) provide client certificate file for mTLS. Default to `(empty)`.
+  > Specifies the path to the certificate file to be used. This is a required parameter if TLS support is enabled.
+- `key-file` (string) provide client private key file for mTLS. Default to `(empty)`.
+  > Specifies the path to the key file corresponding to the certificate file. This is a required parameter if TLS support is enabled.
+- `chan-buffer-size` (int) incoming channel size, number of packet before to drop it. Default to `4096`.
+  > Specifies the maximum number of packets that can be buffered before dropping additional packets.
 
-```yaml
-fluentd:
-  transport: tcp
-  remote-address: 127.0.0.1
-  remote-port: 24224
-  connect-timeout: 5
-  retry-interval: 10
-  flush-interval: 30
-  tag: "dns.collector"
-  tls-insecure: false
-  tls-min-version: 1.2
-  ca-file: ""
-  cert-file: ""
-  key-file: ""
-  buffer-size: 100
-  chan-buffer-size: 65535
-```
diff --git a/docs/performance.md b/docs/performance.md
@@ -17,13 +17,14 @@ The conversion of DNS logs to JSON, text, or PCAP can incur CPU costs. Here's a
 goos: linux
 goarch: amd64
 pkg: github.com/dmachard/go-dnscollector/dnsutils
-cpu: Intel(R) Core(TM) i5-7200U CPU @ 2.50GHz
-BenchmarkDnsMessage_ToTextFormat-4               2600718             460.7 ns/op
-BenchmarkDnsMessage_ToPacketLayer-4              1171467             969.5 ns/op
-BenchmarkDnsMessage_ToDNSTap-4                    993242              1130 ns/op
-BenchmarkDnsMessage_ToExtendedDNSTap-4            618400              1951 ns/op
-BenchmarkDnsMessage_ToJSON-4                      190939              6584 ns/op
-BenchmarkDnsMessage_ToFlatJSON-4                   19868             55533 ns/op
+cpu: Intel(R) Core(TM) i5-7200U CPU @ 2.50GHz 
+BenchmarkDnsMessage_ToTextFormat-4               2555529             450.2 ns/op              80 B/op          4 allocs/op
+BenchmarkDnsMessage_ToPacketLayer-4              1138892             952.0 ns/op            1144 B/op         12 allocs/op
+BenchmarkDnsMessage_ToDNSTap-4                   1036468              1136 ns/op             592 B/op         18 allocs/op
+BenchmarkDnsMessage_ToExtendedDNSTap-4            612438              1970 ns/op            1056 B/op         25 allocs/op
+BenchmarkDnsMessage_ToJSON-4                      188379              6724 ns/op            3632 B/op          3 allocs/op
+BenchmarkDnsMessage_ToFlatten-4                   121525             10151 ns/op            8215 B/op         29 allocs/op
+BenchmarkDnsMessage_ToFlatJSON-4                   20704             58365 ns/op           22104 B/op        220 allocs/op
 ```
 
 ## Memory usage

diff --git a/go.mod b/go.mod
@@ -26,7 +26,6 @@ require (
 	github.com/klauspost/compress v1.17.6
 	github.com/miekg/dns v1.1.58
 	github.com/natefinch/lumberjack v2.0.0+incompatible
-	github.com/nqd/flat v0.2.0
 	github.com/oschwald/maxminddb-golang v1.12.0
 	github.com/prometheus/client_golang v1.18.0
 	github.com/rs/tzsp v0.0.0-20161230003637-8ce729c826b9
@@ -78,7 +77,6 @@ require (
 	github.com/hashicorp/golang-lru v0.6.0 // indirect
 	github.com/hashicorp/memberlist v0.5.0 // indirect
 	github.com/hashicorp/serf v0.10.1 // indirect
-	github.com/imdario/mergo v0.3.15 // indirect
 	github.com/jcmturner/aescts/v2 v2.0.0 // indirect
 	github.com/jcmturner/dnsutils/v2 v2.0.0 // indirect
 	github.com/jcmturner/gofork v1.7.6 // indirect

diff --git a/go.sum b/go.sum
@@ -720,9 +720,6 @@ github.com/hpcloud/tail v1.0.0/go.mod h1:ab1qPbhIpdTxEkNHXyeSf5vhxWSCs/tWer42PpO
 github.com/iancoleman/strcase v0.2.0/go.mod h1:iwCmte+B7n89clKwxIoIXy/HfoL7AsD47ZCWhYzw7ho=
 github.com/ianlancetaylor/demangle v0.0.0-20181102032728-5e5cf60278f6/go.mod h1:aSSvb/t6k1mPoxDqO4vJh6VOCGPwU4O0C2/Eqndh1Sc=
 github.com/ianlancetaylor/demangle v0.0.0-20200824232613-28f6c0f3b639/go.mod h1:aSSvb/t6k1mPoxDqO4vJh6VOCGPwU4O0C2/Eqndh1Sc=
-github.com/imdario/mergo v0.3.12/go.mod h1:jmQim1M+e3UYxmgPu/WyfjB3N3VflVyUjjjwH0dnCYA=
-github.com/imdario/mergo v0.3.15 h1:M8XP7IuFNsqUx6VPK2P9OSmsYsI/YFaGil0uD21V3dM=
-github.com/imdario/mergo v0.3.15/go.mod h1:WBLT9ZmE3lPoWsEzCh9LPo3TiwVN+ZKEjmz+hD27ysY=
 github.com/influxdata/influxdb-client-go v1.4.0 h1:+KavOkwhLClHFfYcJMHHnTL5CZQhXJzOm5IKHI9BqJk=
 github.com/influxdata/influxdb-client-go v1.4.0/go.mod h1:S+oZsPivqbcP1S9ur+T+QqXvrYS3NCZeMQtBoH4D1dw=
 github.com/influxdata/line-protocol v0.0.0-20200327222509-2487e7298839 h1:W9WBk7wlPfJLvMCdtV4zPulc4uCPrlywQOmbFOhgQNU=
@@ -821,8 +818,6 @@ github.com/mwitkow/go-conntrack v0.0.0-20190716064945-2f068394615f/go.mod h1:qRW
 github.com/natefinch/lumberjack v2.0.0+incompatible h1:4QJd3OLAMgj7ph+yZTuX13Ld4UpgHp07nNdFX7mqFfM=
 github.com/natefinch/lumberjack v2.0.0+incompatible/go.mod h1:Wi9p2TTF5DG5oU+6YfsmYQpsTIOm0B1VNzQg9Mw6nPk=
 github.com/niemeyer/pretty v0.0.0-20200227124842-a10e7caefd8e/go.mod h1:zD1mROLANZcx1PVRCS0qkT7pwLkGfwJo4zjcN/Tysno=
-github.com/nqd/flat v0.2.0 h1:g6lXtMxsxrz6PZOO+rNnAJUn/GGRrK4FgVEhy/v+cHI=
-github.com/nqd/flat v0.2.0/go.mod h1:FOuslZmNY082wVfVUUb7qAGWKl8z8Nor9FMg+Xj2Nss=
 github.com/oklog/ulid v1.3.1 h1:EGfNDEx6MqHz8B3uNV6QAib1UR2Lm97sHi3ocA6ESJ4=
 github.com/oklog/ulid v1.3.1/go.mod h1:CirwcVhetQ6Lv90oh/F+FBtV6XMibvdAFo93nm5qn4U=
 github.com/onsi/ginkgo/v2 v2.2.0 h1:3ZNA3L1c5FYDFTTxbFeVGGD8jYvjYauHD30YgLxVsNI=

diff --git a/loggers/fluentd.go b/loggers/fluentd.go
@@ -234,7 +234,7 @@ func (fc *FluentdClient) FlushBuffer(buf *[]dnsutils.DNSMessage) {
 }
 
 func (fc *FluentdClient) Run() {
-	fc.LogInfo("running in background...")
+	fc.LogInfo("waiting dnsmessage to process...")
 
 	// prepare next channels
 	defaultRoutes, defaultNames := fc.RoutingHandler.GetDefaultRoutes()
@@ -246,7 +246,7 @@ func (fc *FluentdClient) Run() {
 	subprocessors := transformers.NewTransforms(&fc.config.OutgoingTransformers, fc.logger, fc.name, listChannel, 0)
 
 	// goroutine to process transformed dns messages
-	go fc.Process()
+	go fc.ProcessDM()
 
 	// init remote conn
 	go fc.ConnectToRemote()
@@ -293,16 +293,16 @@ RUN_LOOP:
 	fc.LogInfo("run terminated")
 }
 
-func (fc *FluentdClient) Process() {
+func (fc *FluentdClient) ProcessDM() {
+	fc.LogInfo("waiting transformed dnsmessage to process...")
+
 	// init buffer
 	bufferDm := []dnsutils.DNSMessage{}
 
 	// init flust timer for buffer
 	flushInterval := time.Duration(fc.config.Loggers.Fluentd.FlushInterval) * time.Second
 	flushTimer := time.NewTimer(flushInterval)
 
-	fc.LogInfo("ready to process")
-
 PROCESS_LOOP:
 	for {
 		select {
@@ -311,7 +311,7 @@ PROCESS_LOOP:
 			break PROCESS_LOOP
 
 		case <-fc.transportReady:
-			fc.LogInfo("connected")
+			fc.LogInfo("connected with remote side")
 			fc.writerReady = true
 
 		// incoming dns message to process

diff --git a/pkgconfig/loggers.go b/pkgconfig/loggers.go
@@ -446,7 +446,7 @@ func (c *ConfigLoggers) SetDefault() {
 	c.Fluentd.KeyFile = ""
 	c.Fluentd.Tag = "dns.collector"
 	c.Fluentd.BufferSize = 100
-	c.Fluentd.ChannelBufferSize = 65535
+	c.Fluentd.ChannelBufferSize = 4096
 
 	c.InfluxDB.Enable = false
 	c.InfluxDB.ServerURL = "http://localhost:8086"

diff --git a/transformers/latency_test.go b/transformers/latency_test.go
@@ -143,6 +143,7 @@ func Test_HashQueries_Expire(t *testing.T) {
 	}
 }
 
+// Bench
 func Benchmark_HashQueries_Set(b *testing.B) {
 	mapexpire := NewHashQueries(10 * time.Second)
-Original file line number
+Diff line change
@@ Expand Up / @@ -143,6 +143,7 @@ func Test_HashQueries_Expire(t *testing.T) { @@
     	}
     }
+    // Bench
     func Benchmark_HashQueries_Set(b *testing.B) {
     	mapexpire := NewHashQueries(10 * time.Second)
@@ Expand Down @@