Major update to support transforms on loggers (#163)

README.md

@@ -8,7 +8,7 @@
 *NOTE: The code before version 1.x is considered beta quality and is subject to breaking changes.*
 
 `DNS-collector` acts as a passive high speed **aggregator, analyzer, transporter and logging** for your DNS messages, written in **Golang**. The DNS traffic can be collected and aggregated from simultaneously sources like DNStap streams, network interface or log files
-and redirect them to several destinations with some transformation (filtering, sampling, privacy, ...).
+and redirect them to several destinations with some transformations on it (filtering, sampling, privacy, ...).
 DNS-collector also contains DNS parser with [`EDNS`](doc/dnsparser.md) support.
 
 ![overview](doc/overview.png)

collectors/dns_processor.go

@@ -83,7 +83,7 @@ func (d *DnsProcessor) Run(sendTo []chan dnsutils.DnsMessage) {
 	d.LogInfo("dns cached enabled: %t", d.cacheSupport)
 
 	// prepare enabled transformers
-	subprocessors := transformers.NewTransforms(d.config, d.logger, d.name)
+	subprocessors := transformers.NewTransforms(&d.config.IngoingTransformers, d.logger, d.name)
 
 	// read incoming dns message
 	d.LogInfo("running... waiting incoming dns message")

collectors/dnstap_processor.go

@@ -129,7 +129,7 @@ func (d *DnstapProcessor) Run(sendTo []chan dnsutils.DnsMessage) {
 	d.LogInfo("dns cached enabled: %t", d.config.Collectors.Dnstap.CacheSupport)
 
 	// prepare enabled transformers
-	subprocessors := transformers.NewTransforms(d.config, d.logger, d.name)
+	subprocessors := transformers.NewTransforms(&d.config.IngoingTransformers, d.logger, d.name)
 
 	// read incoming dns message
 	d.LogInfo("running... waiting incoming dns message")

collectors/powerdns_processor.go

@@ -74,7 +74,7 @@ func (d *PdnsProcessor) Run(sendTo []chan dnsutils.DnsMessage) {
 	pbdm := &powerdns_protobuf.PBDNSMessage{}
 
 	// prepare enabled transformers
-	subprocessors := transformers.NewTransforms(d.config, d.logger, d.name)
+	subprocessors := transformers.NewTransforms(&d.config.IngoingTransformers, d.logger, d.name)
 
 	// read incoming dns message
 	d.LogInfo("running... waiting incoming dns message")

collectors/tail.go

@@ -98,7 +98,7 @@ func (c *Tail) Run() {
 	}
 
 	// prepare enabled transformers
-	subprocessors := transformers.NewTransforms(c.config, c.logger, c.name)
+	subprocessors := transformers.NewTransforms(&c.config.IngoingTransformers, c.logger, c.name)
 
 	// init dns message
 	dm := dnsutils.DnsMessage{}

config.yml

@@ -77,6 +77,9 @@ multiplexer:
     - name: console
       stdout:
         mode: text
+      transforms:
+        normalize:
+          qname-lowercase: true
 
   routes:
     - from: [ tap ]
@@ -408,8 +411,7 @@ multiplexer:
 
 
 ################################################
-# list of transformers
-# transforms:
+# list of transforms to apply on collectors or loggers
 ################################################
 
 # # Use this option to protect user privacy

dnscollector.go

@@ -68,6 +68,7 @@ func main() {
 		// load config
 		cfg := make(map[string]interface{})
 		cfg["loggers"] = output.Params
+		cfg["outgoing-transformers"] = make(map[string]interface{})
 		for _, p := range output.Params {
 			p.(map[string]interface{})["enable"] = true
 		}
@@ -76,6 +77,12 @@ func main() {
 		subcfg := &dnsutils.Config{}
 		subcfg.SetDefault()
 
+		// add transformer
+		for k, v := range output.Transforms {
+			v.(map[string]interface{})["enable"] = true
+			cfg["outgoing-transformers"].(map[string]interface{})[k] = v
+		}
+
 		// copy global config
 		subcfg.Global = config.Global
 
@@ -132,7 +139,7 @@ func main() {
 		// load config
 		cfg := make(map[string]interface{})
 		cfg["collectors"] = input.Params
-		cfg["transformers"] = make(map[string]interface{})
+		cfg["ingoing-transformers"] = make(map[string]interface{})
 		for _, p := range input.Params {
 			p.(map[string]interface{})["enable"] = true
 		}
@@ -144,7 +151,7 @@ func main() {
 		// add transformer
 		for k, v := range input.Transforms {
 			v.(map[string]interface{})["enable"] = true
-			cfg["transformers"].(map[string]interface{})[k] = v
+			cfg["ingoing-transformers"].(map[string]interface{})[k] = v
 		}
 
 		// copy global config
@@ -172,6 +179,7 @@ func main() {
 		}
 	}
 
+	// here the multiplexer logic
 	// connect collectors between loggers
 	for _, routes := range config.Multiplexer.Routes {
 		var logwrks []dnsutils.Worker

dnsutils/config.go

@@ -28,12 +28,6 @@ func IsValidTLS(mode string) bool {
 	return false
 }
 
-type MultiplexTransformers struct {
-	Name       string                 `yaml:"naame"`
-	Transforms map[string]interface{} `yaml:",inline"`
-	Params     map[string]interface{} `yaml:",inline"`
-}
-
 type MultiplexInOut struct {
 	Name       string                 `yaml:"name"`
 	Transforms map[string]interface{} `yaml:"transforms"`
@@ -45,6 +39,81 @@ type MultiplexRoutes struct {
 	Dst []string `yaml:"to,flow"`
 }
 
+type ConfigTransformers struct {
+	UserPrivacy struct {
+		Enable        bool `yaml:"enable"`
+		AnonymizeIP   bool `yaml:"anonymize-ip"`
+		MinimazeQname bool `yaml:"minimaze-qname"`
+	} `yaml:"user-privacy"`
+	Normalize struct {
+		Enable         bool `yaml:"enable"`
+		QnameLowerCase bool `yaml:"qname-lowercase"`
+	} `yaml:"normalize"`
+	Filtering struct {
+		Enable          bool     `yaml:"enable"`
+		DropFqdnFile    string   `yaml:"drop-fqdn-file"`
+		DropDomainFile  string   `yaml:"drop-domain-file"`
+		KeepFqdnFile    string   `yaml:"keep-fqdn-file"`
+		KeepDomainFile  string   `yaml:"keep-domain-file"`
+		DropQueryIpFile string   `yaml:"drop-queryip-file"`
+		KeepQueryIpFile string   `yaml:"keep-queryip-file"`
+		DropRcodes      []string `yaml:"drop-rcodes,flow"`
+		LogQueries      bool     `yaml:"log-queries"`
+		LogReplies      bool     `yaml:"log-replies"`
+		Downsample      int      `yaml:"downsample"`
+	} `yaml:"filtering"`
+	GeoIP struct {
+		Enable        bool   `yaml:"enable"`
+		DbCountryFile string `yaml:"mmdb-country-file"`
+		DbCityFile    string `yaml:"mmdb-city-file"`
+		DbAsnFile     string `yaml:"mmdb-asn-file"`
+	} `yaml:"geoip"`
+	Suspicious struct {
+		Enable             bool     `yaml:"enable"`
+		ThresholdQnameLen  int      `yaml:"threshold-qname-len"`
+		ThresholdPacketLen int      `yaml:"threshold-packet-len"`
+		ThresholdSlow      float64  `yaml:"threshold-slow"`
+		CommonQtypes       []string `yaml:"common-qtypes,flow"`
+		UnallowedChars     []string `yaml:"unallowed-chars,flow"`
+		ThresholdMaxLabels int      `yaml:"threshold-max-labels"`
+	} `yaml:"suspicious"`
+}
+
+func (c *ConfigTransformers) SetDefault() {
+	c.Suspicious.Enable = false
+	c.Suspicious.ThresholdQnameLen = 100
+	c.Suspicious.ThresholdPacketLen = 1000
+	c.Suspicious.ThresholdSlow = 1.0
+	c.Suspicious.CommonQtypes = []string{"A", "AAAA", "TXT", "CNAME", "PTR",
+		"NAPTR", "DNSKEY", "SRV", "SOA", "NS", "MX", "DS"}
+	c.Suspicious.UnallowedChars = []string{"\"", "==", "/", ":"}
+	c.Suspicious.ThresholdMaxLabels = 10
+
+	c.UserPrivacy.Enable = false
+	c.UserPrivacy.AnonymizeIP = false
+	c.UserPrivacy.MinimazeQname = false
+
+	c.Normalize.Enable = false
+	c.Normalize.QnameLowerCase = false
+
+	c.Filtering.Enable = false
+	c.Filtering.DropFqdnFile = ""
+	c.Filtering.DropDomainFile = ""
+	c.Filtering.KeepFqdnFile = ""
+	c.Filtering.KeepDomainFile = ""
+	c.Filtering.DropQueryIpFile = ""
+	c.Filtering.DropRcodes = []string{}
+	c.Filtering.LogQueries = true
+	c.Filtering.LogReplies = true
+	c.Filtering.Downsample = 0
+
+	c.GeoIP.Enable = false
+	c.GeoIP.DbCountryFile = ""
+	c.GeoIP.DbCityFile = ""
+	c.GeoIP.DbAsnFile = ""
+}
+
+/* main configuration */
 type Config struct {
 	Global struct {
 		TextFormat string `yaml:"text-format"`
@@ -108,45 +177,7 @@ type Config struct {
 		} `yaml:"pcap"`
 	} `yaml:"collectors"`
 
-	Transformers struct {
-		UserPrivacy struct {
-			Enable        bool `yaml:"enable"`
-			AnonymizeIP   bool `yaml:"anonymize-ip"`
-			MinimazeQname bool `yaml:"minimaze-qname"`
-		} `yaml:"user-privacy"`
-		Normalize struct {
-			Enable         bool `yaml:"enable"`
-			QnameLowerCase bool `yaml:"qname-lowercase"`
-		} `yaml:"normalize"`
-		Filtering struct {
-			Enable          bool     `yaml:"enable"`
-			DropFqdnFile    string   `yaml:"drop-fqdn-file"`
-			DropDomainFile  string   `yaml:"drop-domain-file"`
-			KeepFqdnFile    string   `yaml:"keep-fqdn-file"`
-			KeepDomainFile  string   `yaml:"keep-domain-file"`
-			DropQueryIpFile string   `yaml:"drop-queryip-file"`
-			KeepQueryIpFile string   `yaml:"keep-queryip-file"`
-			DropRcodes      []string `yaml:"drop-rcodes,flow"`
-			LogQueries      bool     `yaml:"log-queries"`
-			LogReplies      bool     `yaml:"log-replies"`
-			Downsample      int      `yaml:"downsample"`
-		} `yaml:"filtering"`
-		GeoIP struct {
-			Enable        bool   `yaml:"enable"`
-			DbCountryFile string `yaml:"mmdb-country-file"`
-			DbCityFile    string `yaml:"mmdb-city-file"`
-			DbAsnFile     string `yaml:"mmdb-asn-file"`
-		} `yaml:"geoip"`
-		Suspicious struct {
-			Enable             bool     `yaml:"enable"`
-			ThresholdQnameLen  int      `yaml:"threshold-qname-len"`
-			ThresholdPacketLen int      `yaml:"threshold-packet-len"`
-			ThresholdSlow      float64  `yaml:"threshold-slow"`
-			CommonQtypes       []string `yaml:"common-qtypes,flow"`
-			UnallowedChars     []string `yaml:"unallowed-chars,flow"`
-			ThresholdMaxLabels int      `yaml:"threshold-max-labels"`
-		} `yaml:"suspicious"`
-	} `yaml:"transformers"`
+	IngoingTransformers ConfigTransformers `yaml:"ingoing-transformers"`
 
 	Loggers struct {
 		Stdout struct {
@@ -294,6 +325,8 @@ type Config struct {
 		} `yaml:"elasticsearch"`
 	} `yaml:"loggers"`
 
+	OutgoingTransformers ConfigTransformers `yaml:"outgoing-transformers"`
+
 	Multiplexer struct {
 		Collectors []MultiplexInOut  `yaml:"collectors"`
 		Loggers    []MultiplexInOut  `yaml:"loggers"`
@@ -360,38 +393,8 @@ func (c *Config) SetDefault() {
 	c.Collectors.IngestPcap.DropReplies = false
 	c.Collectors.IngestPcap.DeleteAfter = false
 
-	// Transformers
-	c.Transformers.Suspicious.Enable = false
-	c.Transformers.Suspicious.ThresholdQnameLen = 100
-	c.Transformers.Suspicious.ThresholdPacketLen = 1000
-	c.Transformers.Suspicious.ThresholdSlow = 1.0
-	c.Transformers.Suspicious.CommonQtypes = []string{"A", "AAAA", "TXT", "CNAME", "PTR",
-		"NAPTR", "DNSKEY", "SRV", "SOA", "NS", "MX", "DS"}
-	c.Transformers.Suspicious.UnallowedChars = []string{"\"", "==", "/", ":"}
-	c.Transformers.Suspicious.ThresholdMaxLabels = 10
-
-	c.Transformers.UserPrivacy.Enable = false
-	c.Transformers.UserPrivacy.AnonymizeIP = false
-	c.Transformers.UserPrivacy.MinimazeQname = false
-
-	c.Transformers.Normalize.Enable = false
-	c.Transformers.Normalize.QnameLowerCase = false
-
-	c.Transformers.Filtering.Enable = false
-	c.Transformers.Filtering.DropFqdnFile = ""
-	c.Transformers.Filtering.DropDomainFile = ""
-	c.Transformers.Filtering.KeepFqdnFile = ""
-	c.Transformers.Filtering.KeepDomainFile = ""
-	c.Transformers.Filtering.DropQueryIpFile = ""
-	c.Transformers.Filtering.DropRcodes = []string{}
-	c.Transformers.Filtering.LogQueries = true
-	c.Transformers.Filtering.LogReplies = true
-	c.Transformers.Filtering.Downsample = 0
-
-	c.Transformers.GeoIP.Enable = false
-	c.Transformers.GeoIP.DbCountryFile = ""
-	c.Transformers.GeoIP.DbCityFile = ""
-	c.Transformers.GeoIP.DbAsnFile = ""
+	// Transformers for collectors
+	c.IngoingTransformers.SetDefault()
 
 	// Loggers
 	c.Loggers.Stdout.Enable = false
@@ -523,6 +526,10 @@ func (c *Config) SetDefault() {
 
 	c.Loggers.ElasticSearchClient.Enable = false
 	c.Loggers.ElasticSearchClient.URL = ""
+
+	// Transformers for loggers
+	c.OutgoingTransformers.SetDefault()
+
 }
 
 func (c *Config) GetServerIdentity() string {
@@ -583,3 +590,9 @@ func GetFakeConfig() *Config {
 	config.SetDefault()
 	return config
 }
+
+func GetFakeConfigTransformers() *ConfigTransformers {
+	config := &ConfigTransformers{}
+	config.SetDefault()
+	return config
+}

doc/configuration.md

@@ -161,7 +161,7 @@ multiplexer:
 ```
 ## Transformers
 
-Some transformations can be done after the collect.
+Some transformations can be done on collectors or loggers.
 
 ### Normalize
 

doc/overview.drawio

@@ -1 +1 @@
-<mxfile host="app.diagrams.net" modified="2022-06-22T18:06:44.738Z" agent="5.0 (X11)" etag="TcZ8aJD9gEtu3iOVryzu" version="20.0.1" type="device"><diagram id="ufO1G3qEnvI_ADMoatvI" name="Page-1">5Vxte5o6GP41fpQLkhDCR61tt562607Xres3KlHZkDjEqfv1J1RQTAJiBdEze1294OHh7c79vCbaghfjxXXoTEZ3zKV+C+juogV7LQBsYvD/sWC5EpgGWgmGoeeuRMZG8Oj9oYlQT6Qzz6XTLcWIMT/yJtvCPgsC2o+2ZE4Ysvm22oD523edOEMqCR77ji9Lv3luNFpJialv5B+oNxyldzb05MjYSZUTwXTkuGyeEcHLFrwIGYtWW+PFBfVj7FJcVudd5RxdP1hIg6jMCd0/5pcxmL5Onr/dB8TpXP+wL9rJVX47/ix54eRho2WKQMhmgUvji+gt2J2PvIg+Tpx+fHTOh5zLRtHY53sG35xGIftJL5jPQi4JWMDVugPP91NRC8CBGf/FchZEGfnqw+Xym6WPScOILjKi5E2vKRvTKFxyleQoQAnqS2F/vhlEM+XYKDOAwE6ETkKc4fraG2z5RgLvHlDDfaA2dkMtoOo6lAz6eUPghP3EtvSKAMb2NsAm0EwJYkMJMbZS1cpBRgqQsc9v3H3lG8N4gyPjc0/Bwml6iN9pfVQaE45HtCfHU8x9bxjw3T5HmXJ5N0bX466lkxwYe64b30Y50ttmV8GIQR2VGDGkGrC6TMKs1yQGgwHoH88ksCH4HNPSgMIm4JFtAu+2iVs2HNK/zSCwCTXLLDNkRzUKSxouN5hGHILqA3MVcYBgTcCQyABaa1eThRDWxXgiQfgYOZE35WSbniaMGCENEgFHfU3PLJQKLkJ7zdrKsbQlLLmzuPI4eucCJLKIZpcDEhmaXheQhpxr3/ELnxMloWUeGUmzjV7neG68fOiPL5/w89eegdrpU2UQoy6v2pJdFkYjNmSB419upEL42OjcMjZJkPxBo2iZ5ATOLGIqnNMikAPXdZ3paJ2axDsPTsQDW/AmAbqxHof46bZGQTEwUzYL+7SIP0lyFDnhkEZF8JjqcQ2pz53g7+0nUQ3O26mdMHSWGYUJ84JoKo3d+voHGAaQDKN3whFP5WEAD4KonF1ADeGaPMyaSke1C45kuHxOzn/b+R7vaGa621tkD/aWyV5eV2B73PgR/PZRWmApa1IaSdpLyhhTUTYmWPChxjVePMTmtCGVAbZrB6gL6eXqhZKzNjzJt9Kc+2DxPih7uRL6QKDp6gnUZwNTONsC22+1gl96q3d4FuUgI+ucDOKIAaYNly//YraYfLu78RZ3bm9501GZhDq+HGoCZZ1c0VNmgoWmaWcTKSCwVQWmMlJs0v9DQPS+fBzd9Bberb40rY7/7AznC0VSn5b+rvc7rfx9NuQ6g7dUf90XyBw/TdDlghTatqZnP1jGX12fVtKTMT/dvZA7rCP29P3j7ezlt+NY58RiGVCE5WZ6nQh+/3I/uv1JPv/oBPPuza+Lf66fJmlX67iuXWowUqxoMMbdeMt+1Us49VLeWvn+sGQCk2bXFWcw+yYeWBc4ZAizVbv0EclPPN6RLygxbSRdEOdxTEpcpOIUAa/wkDz4IBrZJ8kiaO3JItuun0WqiT4puN6y4TSGi80iLxjmBVjptPEsnvr26YKGdQRl0b+RPs2fQKkgtiCxQECWahZRV6RHxKopsKAmnEBezQB2FQ104UXP6UX4dqbU4Hubk+Kd/EJjfx9UyuMUVdRZj1NkRg17HCR6HB0WehxJH1frcRbz3tX9kl312+Tzp+tZu7P4SpWrOCTXkc4ilfQ0kb+eDnwNRb0zLQQQ0AyrbOpaQeKqHCzQiH85ZTdRxOmGzR/aQoDCxQmHpG8VmL98thgODSDWTzl9QNkPSa03e/tC1bXelN7bPBeWN1PSFXVlziQyilRHBilkpKSv661c06gqMjaSzDVDqaI+05lQKqXEmiKk2NtK+qCAUjIhhTWia5e5t6/FoiUIF6rO1ypJ3oivbXb6Dsi8PmIWoZgmEzJNEmeam46zUYpYu6fc5Rk2yxKTXDlnqJl+ZzXJ1ixtFd22okygdtpaWLN127YRBpjYRlLbbiZOSMn0c+9paETEpZfISJmcOxdd4qRdE9L8hYHYQsT12Yt6VrolV+uHFNAtdWGsYGhurWxAJOKSKYsztTJRrT6soBNXtEajuK3xwOY07N0/lm5suMHU9abRcZobVYwNkMeGwGZHZq+F+41RGEJ5nvKoMAEJpqfgNcalebRkUuV0x44KmLz2vXmkoBgUoaaTZmGSZ5i63kmQahsqpDfsptI67G9LTfddDqYET9GNym8x1FBR8WRPWGVrG5ph2puPQJuKclMgLMVE2y3cffV3ZqQ8g4V7vWfN6an85ZnmHYsc2xFu3g3Lq+gOXsRVT2hHVtOOWObQWa2s2dehHrTUhlThUPf1eqa130SXpF/UzZfPFpeVk3e2XrEZL7GVGm3rlNaW1tDW7D0NeAJMr2rG4S+gPSmclNqpD41DaC9NFFRDewRMTYcIEWCZBAP9uPkDlNs2j17w88BvOVYQGC1b8A6Q51kwsy684XYXVHVVMu2rDHb41yz+2Zi3X1BpT9+8QocrGMZksWpkJcfF1lasf9CF3GDa7qc/oFHYQlvd6qQ7ayIhEODhouEOCGqkZD3iV4/U1FcsCqwlSBxmoHJ76mIW+ifHY0AMzUKZAlL+aupxOd3IKoymOY1ASU4bx/o+XeFjZkj9EMbXHtHZCcZsy9Lshst+1Mzatmq6iudSBB9sFO8qB7Curkvzv18i6Nu4WD91hWr9/2HVzHc3P3G4Ut/8TiS8/A8=</diagram></mxfile>
+<mxfile host="app.diagrams.net" modified="2022-11-13T06:31:27.699Z" agent="5.0 (X11)" etag="XESKpodHZSGLUGk0bACG" version="20.0.2" type="device"><diagram id="ufO1G3qEnvI_ADMoatvI" name="Page-1">7Vxte5o6GP41XtfOB7mAhAAftbbdetquXdut6zcqUdmQOMSq+/UnKCgk4cUKottxH0aehLc79/NK0hY4Gy8ufWsyuiE2dluqbC9aoNdSVUUxNPpfKFmuJUhV1oKh79jRoK3gwfmNI6EcSWeOjaepgQEhbuBM0sI+8TzcD1Iyy/fJPD1sQNz0XSfWEHOCh77l8tJvjh2M1lJDk7fyj9gZjuI7K3LUM7biwZFgOrJsMk+IwHkLnPmEBOuj8eIMuyF4MS7r8y4yejcP5mMvKHNC97f2OFanr5Pnb7eeYXUuf5hn7egqb5Y7i144ethgGSPgk5ln4/Aicgt05yMnwA8Tqx/2zumcU9koGLu0pdDDaeCTn/iMuMSnEo94dFh34LhuLGqpYKCF/0I58YKEfP2jcv7N4sfEfoAXCVH0ppeYjHHgL+mQqFeFEepLpj3fTqIWc2yUmEDVjIRWRJzh5tpbbOlBBO8OUINdoFaKoWZQtS1sDPpZU2D5/Ui35IoARmYaYE2VNA5iJb5wCmKkx0MrBxkKQEYuvXH3lR4MwwOKjEstBfGncRe906aXmxOKR7Ajx2PMXWfo0WafooypvBui61DT0ok6xo5th7cRznRa7SqYMSDDEjMGRTpRl0po9arEYDBQ+4dTCaQwNkfTJVWgE0AEcY06gYp14poMh/hvUwikAUnXykzZQZVC56bL9qYBhaB6x1yFHzCQxGBo8K5W35iaJISwLghNDkJK8AuHAneUGCIIJWCkQATUHsjJH+IxFZASalJsQysHVeFjxRt6Yac/PRlUhdQUwahKcl0wxi+SgAvbNOWImsQPRmRIPMs930oZ27cdc03IJILxBw6CZeTQrFlA0iBTGP3lc3T+qvE9bITGLmr3Fsne3jJqZcXu6VmjPWj12/TEyZKaN5mB5Q9xUGwLQ4Byp9zHrhU4b+mcTTR10al3xKGPsjVijOcGMmOZpmTm93F01pYAHd+3lolhk3DAtPx94mRjy6f1FTPO1pin1BmOruHknnJF0w1YpZirteHrHM2Vl4/98fkTev7aU2Ab6sfE3CLi8jS0reloE0mGjTsroHGIt5KospLH1DUB1qI2WL58QWQx+XZz5Sxu7N7yqtMWMFoIIlBqofSuTERGmktQh0nK7Do+zdx3EM55/DS66i2ca3mp6R332RrOFwI3HgeotvMWx6eB5biJuDXRc5xuiQ+YgK5Jppbw9XwAKo6fQCU5g/b55sW4QTIkT98/Xc9e3ixLF9SFJEk6FUAhOmQE+v3xdnT907j/0fHm3atfZ/9ePk3inOuwppJLfzESpL9hrUg3X+USRrKUORS+Pyjp4M2jsIaaqacJJDOl1ILxYCc/jnRTAmbil743y8nqvHreTOUb2ZtZWHF38YKm7Rm2ljvnC5kFjjfc1Bp8dmyV1pqlvtHH2ZWfCowOZGMxqIvKn7IguTD0mmwOPKbwrDCxwAsneI4vQo8TUR1tbU8KG9kxHVsE17BhQ5HJM9RXsEpPStm0vGQkadPyNKphmwZlJsoosGnc+KjAX1WEJ4RKa4Sy72BeM641Lyo+FRqyqbWp5dMwf/zeNFzMexe3S3LRbxv3ny9n7c7iK27GcjZDqbxo/1QoJTO5qFFk2Zjx6i7RGmC+JG88f0FtiH8MxFxIqS3QE5K8EVtbRfkwK8TQi0KMcgohqtKovEI0WZ9UEOObDVVS9G3BQCnFyGw9y+SsqutssUJR2aJDzbxtpOpYa9m7kH6C7DkvFKidfjqSTJkmqhCpyDAVGTAu2mApUVW9XGMtt5JPV3a8vFN9nb6mypbmUH1sFxfZW3w6vk9iXFD0zuVp+XRYAZCFLlFxS2TEhiAhBhUkxEIsRYvbuDLFHZljv3f7ULquYXtT25kGh6lrHGj6VH76DNDs5O20Nuh4kOQVAQC+Gn1QJFUOySfvNYTuJADlqQlgGP00iym/SOckwASs5waSbDSLJF9/7jqnQs00mlBu2GTGCeVxVIAbXVlSGGILqmh5pZH6MzxD0tOLlaCpSIqW+EzE0KaqGBvsVk0pGF8YY9OEEuz0njUH3MaJWnLOLULUvDHnV2zsvWygsTAD6g2bc8DnLw+O93PP5Zb7rIIqj6fOrhfStKL1rAfFFjZS8q9vTZr4JVXeyeUu7W24sM+SBkQsyXJFBeP3/laUC2pCK+/88KojPDtJ1VSNI1PNZj4KVxPFlliA8M4odq9lVkei4JrOKGxBrMmNl5VWpoLzZyPx3Xb+coe0cCcB97llU44wN9uIqo9ZP9/rY/VLu6N+nP8av5A7e941BRWdVJ0yoTro1yzcX7zaatuersjfoQPMyWJdsIy62RJmOHyf6zz6ljcdEH+c3FK2vmh+rfSo3uLDne+8WX3KcDpNMo1k/8kt+pZ5P9awua4zmeJi1/CO9W2cRxCYmEwnARST0TxTUM1HAr9QxZJaIemP4vv13os03utnhPa7MqeRvaXh4C7CZLIw9q8aFI3Xc5atFboIyO7q3MlF6JlrejWjLv+QtxnlFFbvC3bp0VyRszTCXXqVbH8Q2hpRef+oXNO7HeypOCBuw5EBN1FWUx4orq//gbQ4qrf4QMlG897V5oG/LvLiiB+vXmyM9QXLKJrnyx9vDNlo/CiM4Z+bhJ4sLRo3FQWbyf7nxMHdR62JO21u/37dOnnZ/hVAcP4f</diagram></mxfile>

loggers/dnstap.go

@@ -8,6 +8,7 @@ import (
 	"time"
 
 	"github.com/dmachard/go-dnscollector/dnsutils"
+	"github.com/dmachard/go-dnscollector/transformers"
 	"github.com/dmachard/go-dnstap-protobuf"
 	"github.com/dmachard/go-framestream"
 	"github.com/dmachard/go-logger"
@@ -83,6 +84,9 @@ func (o *DnstapSender) Stop() {
 func (o *DnstapSender) Run() {
 	o.LogInfo("running in background...")
 
+	// prepare transforms
+	subprocessors := transformers.NewTransforms(&o.config.OutgoingTransformers, o.logger, o.name)
+
 	dt := &dnstap.Dnstap{}
 	frame := &framestream.Frame{}
 
@@ -147,6 +151,11 @@ LOOP:
 						select {
 						case dm := <-o.channel:
 
+							// apply tranforms
+							if subprocessors.ProcessMessage(&dm) == transformers.RETURN_DROP {
+								continue
+							}
+
 							dt.Reset()
 
 							t := dnstap.Dnstap_MESSAGE
@@ -240,6 +249,11 @@ LOOP:
 		o.LogInfo("closing tcp connection")
 		o.conn.Close()
 	}
+
 	o.LogInfo("run terminated")
+
+	// cleanup transformers
+	subprocessors.Reset()
+
 	o.done <- true
 }