add docs

.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/

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-10-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)
 

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:

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

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

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

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 ]

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

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

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
-```

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

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"

transformers/subprocessors_test.go

@@ -16,6 +16,35 @@ const (
 	Localhost   = "localhost"
 )
 
+// Bench to init DNS message
+func BenchmarkTransforms_Init(b *testing.B) {
+	config := pkgconfig.GetFakeConfigTransformers()
+	config.Suspicious.Enable = true
+	config.GeoIP.Enable = true
+	config.GeoIP.DBCountryFile = "../testsdata/GeoLite2-Country.mmdb"
+	config.GeoIP.Enable = true
+	config.GeoIP.DBASNFile = "../testsdata/GeoLite2-ASN.mmdb"
+	config.UserPrivacy.Enable = true
+	config.UserPrivacy.MinimazeQname = true
+	config.UserPrivacy.Enable = true
+	config.UserPrivacy.AnonymizeIP = true
+	config.Normalize.Enable = true
+	config.Normalize.QnameLowerCase = true
+	config.Filtering.Enable = true
+	config.Filtering.KeepDomainFile = "../testsdata/filtering_keep_domains.txt"
+
+	channels := []chan dnsutils.DNSMessage{}
+	subprocessors := NewTransforms(config, logger.New(false), "test", channels, 0)
+
+	dm := dnsutils.GetFakeDNSMessage()
+
+	b.ResetTimer()
+	for i := 0; i < b.N; i++ {
+		subprocessors.InitDNSMessageFormat(&dm)
+	}
+}
+
+// Other tests
 func TestTransformsSuspicious(t *testing.T) {
 	// config
 	config := pkgconfig.GetFakeConfigTransformers()