config: use a validator for better configuration validation

cmd/config.go

@@ -14,9 +14,11 @@ import (
 	"strconv"
 	"strings"
 
+	"github.com/go-playground/validator/v10"
 	"github.com/mitchellh/mapstructure"
 	"gopkg.in/yaml.v2"
 
+	"akvorado/common/helpers"
 	"akvorado/inlet/flow"
 	"akvorado/orchestrator/clickhouse"
 )
@@ -131,10 +133,18 @@ func (c ConfigRelatedOptions) Parse(out io.Writer, component string, config inte
 		}
 	}
 
-	// Dump configuration if requested
+	// Validate and dump configuration if requested
 	if c.BeforeDump != nil {
 		c.BeforeDump()
 	}
+	if err := helpers.Validate.Struct(config); err != nil {
+		switch verr := err.(type) {
+		case validator.ValidationErrors:
+			return fmt.Errorf("invalid configuration:\n%w", verr)
+		default:
+			return fmt.Errorf("unexpected internal error: %w", verr)
+		}
+	}
 	if c.Dump {
 		output, err := yaml.Marshal(config)
 		if err != nil {

cmd/config_test.go

@@ -26,9 +26,9 @@ type dummyConfiguration struct {
 	Module2 dummyModule2Configuration
 }
 type dummyModule1Configuration struct {
-	Listen  string
-	Topic   string
-	Workers int
+	Listen  string `validate:"listen"`
+	Topic   string `validate:"gte=3"`
+	Workers int    `validate:"gte=1"`
 }
 type dummyModule2Configuration struct {
 	Details     dummyModule2DetailsConfiguration
@@ -66,6 +66,30 @@ func dummyDefaultConfiguration() dummyConfiguration {
 	}
 }
 
+func TestValidation(t *testing.T) {
+	config := `---
+module1:
+ topic: fl
+ workers: -5
+`
+	configFile := filepath.Join(t.TempDir(), "config.yaml")
+	ioutil.WriteFile(configFile, []byte(config), 0644)
+
+	c := cmd.ConfigRelatedOptions{
+		Path: configFile,
+	}
+
+	parsed := dummyDefaultConfiguration()
+	out := bytes.NewBuffer([]byte{})
+	if err := c.Parse(out, "dummy", &parsed); err == nil {
+		t.Fatal("Parse() didn't error")
+	} else if diff := helpers.Diff(err.Error(), `invalid configuration:
+Key: 'dummyConfiguration.Module1.Topic' Error:Field validation for 'Topic' failed on the 'gte' tag
+Key: 'dummyConfiguration.Module1.Workers' Error:Field validation for 'Workers' failed on the 'gte' tag`); diff != "" {
+		t.Fatalf("Parse() (-got, +want):\n%s", diff)
+	}
+}
+
 func TestDump(t *testing.T) {
 	// Configuration file
 	config := `---

common/clickhousedb/config.go

@@ -10,17 +10,17 @@ import (
 // Configuration defines how we connect to a Clickhouse database
 type Configuration struct {
 	// Servers define the list of clickhouse servers to connect to (with ports)
-	Servers []string
+	Servers []string `validate:"min=1,dive,listen"`
 	// Database defines the database to use
-	Database string
+	Database string `validate:"required"`
 	// Username defines the username to use for authentication
-	Username string
+	Username string `validate:"required"`
 	// Password defines the password to use for authentication
 	Password string
 	// MaxOpenConns tells how many parallel connections to ClickHouse we want
-	MaxOpenConns int
+	MaxOpenConns int `validate:"min=1"`
 	// DialTimeout tells how much time to wait when connecting to ClickHouse
-	DialTimeout time.Duration
+	DialTimeout time.Duration `validate:"min=100ms"`
 }
 
 // DefaultConfiguration represents the default configuration for connecting to Clickhouse

common/clickhousedb/config_test.go

@@ -0,0 +1,16 @@
+// SPDX-FileCopyrightText: 2022 Free Mobile
+// SPDX-License-Identifier: AGPL-3.0-only
+
+package clickhousedb
+
+import (
+	"testing"
+
+	"akvorado/common/helpers"
+)
+
+func TestDefaultConfiguration(t *testing.T) {
+	if err := helpers.Validate.Struct(DefaultConfiguration()); err != nil {
+		t.Fatalf("validate.Struct() error:\n%+v", err)
+	}
+}

common/helpers/validator.go

@@ -0,0 +1,38 @@
+// SPDX-FileCopyrightText: 2022 Free Mobile
+// SPDX-License-Identifier: AGPL-3.0-only
+
+package helpers
+
+import (
+	"net"
+	"strconv"
+
+	"github.com/go-playground/validator/v10"
+)
+
+// Validate is a validator instance to be used everywhere.
+var Validate *validator.Validate
+
+// isListen validates a <dns>:<port> combination for fields typically used for listening address
+func isListen(fl validator.FieldLevel) bool {
+	val := fl.Field().String()
+	host, port, err := net.SplitHostPort(val)
+	if err != nil {
+		return false
+	}
+	// Port must be a iny <= 65535.
+	if portNum, err := strconv.ParseInt(port, 10, 32); err != nil || portNum > 65535 || portNum < 0 {
+		return false
+	}
+
+	// If host is specified, it should match a DNS name
+	if host != "" {
+		return Validate.Var(host, "hostname_rfc1123") == nil
+	}
+	return true
+}
+
+func init() {
+	Validate = validator.New()
+	Validate.RegisterValidation("listen", isListen)
+}

common/helpers/validator_test.go

@@ -0,0 +1,39 @@
+// SPDX-FileCopyrightText: 2022 Free Mobile
+// SPDX-License-Identifier: AGPL-3.0-only
+
+package helpers_test
+
+import (
+	"testing"
+
+	"akvorado/common/helpers"
+)
+
+func TestListenValidator(t *testing.T) {
+	s := struct {
+		Listen string `validate:"listen"`
+	}{}
+	cases := []struct {
+		Listen string
+		Err    bool
+	}{
+		{"127.0.0.1:161", false},
+		{"localhost:161", false},
+		{"0.0.0.0:161", false},
+		{"0.0.0.0:0", false},
+		{"127.0.0.1:0", false},
+		{"localhost", true},
+		{"127.0.0.1", true},
+		{"127.0.0.1:what", true},
+		{"127.0.0.1:100000", true},
+	}
+	for _, tc := range cases {
+		s.Listen = tc.Listen
+		err := helpers.Validate.Struct(s)
+		if err == nil && tc.Err {
+			t.Error("Validate.Struct() expected an error")
+		} else if err != nil && !tc.Err {
+			t.Errorf("Validate.Struct() error:\n%+v", err)
+		}
+	}
+}

common/http/config.go

@@ -6,7 +6,7 @@ package http
 // Configuration describes the configuration for the HTTP server.
 type Configuration struct {
 	// Listen defines the listening string to listen to.
-	Listen string
+	Listen string `validate:"listen"`
 	// Profiler enables Go profiler as /debug
 	Profiler bool
 }

common/http/config_test.go

@@ -0,0 +1,16 @@
+// SPDX-FileCopyrightText: 2022 Free Mobile
+// SPDX-License-Identifier: AGPL-3.0-only
+
+package http
+
+import (
+	"testing"
+
+	"akvorado/common/helpers"
+)
+
+func TestDefaultConfiguration(t *testing.T) {
+	if err := helpers.Validate.Struct(DefaultConfiguration()); err != nil {
+		t.Fatalf("validate.Struct() error:\n%+v", err)
+	}
+}

common/kafka/config.go

@@ -8,9 +8,9 @@ import "github.com/Shopify/sarama"
 // Configuration defines how we connect to a Kafka cluster.
 type Configuration struct {
 	// Topic defines the topic to write flows to.
-	Topic string
+	Topic string `validate:"required"`
 	// Brokers is the list of brokers to connect to.
-	Brokers []string
+	Brokers []string `min=1,dive,validate:"listen"`
 	// Version is the version of Kafka we assume to work
 	Version Version
 }

common/kafka/config_test.go

@@ -0,0 +1,16 @@
+// SPDX-FileCopyrightText: 2022 Free Mobile
+// SPDX-License-Identifier: AGPL-3.0-only
+
+package kafka
+
+import (
+	"testing"
+
+	"akvorado/common/helpers"
+)
+
+func TestDefaultConfiguration(t *testing.T) {
+	if err := helpers.Validate.Struct(DefaultConfiguration()); err != nil {
+		t.Fatalf("validate.Struct() error:\n%+v", err)
+	}
+}

console/data/docs/06-internals.md

@@ -221,6 +221,9 @@ spawned by the other components and wait for signals to terminate. If
  - [github.com/eapache/go-resiliency](https://github.com/eapache/go-resiliency)
    implements several resiliency pattersn, including the breaker
    pattern.
+ - [github.com/go-playground/validator](https://github.com/go-playground/validator)
+   implements struct validation using tags. We use it to had better
+   validation on configuration structures.
 
 ## Future plans
 

console/database/config.go

@@ -6,9 +6,9 @@ package database
 // Configuration describes the configuration for the authentication component.
 type Configuration struct {
 	// Driver defines the driver for the database
-	Driver string
+	Driver string `validate:"required"`
 	// DSN defines the DSN to connect to the database
-	DSN string
+	DSN string `validate:"required"`
 }
 
 // DefaultConfiguration represents the default configuration for the console component.

console/database/config_test.go

@@ -0,0 +1,16 @@
+// SPDX-FileCopyrightText: 2022 Free Mobile
+// SPDX-License-Identifier: AGPL-3.0-only
+
+package database
+
+import (
+	"testing"
+
+	"akvorado/common/helpers"
+)
+
+func TestDefaultConfiguration(t *testing.T) {
+	if err := helpers.Validate.Struct(DefaultConfiguration()); err != nil {
+		t.Fatalf("validate.Struct() error:\n%+v", err)
+	}
+}

inlet/core/config.go

@@ -6,7 +6,7 @@ package core
 // Configuration describes the configuration for the core component.
 type Configuration struct {
 	// Number of workers for the core component
-	Workers int
+	Workers int `validate:"min=1"`
 	// ExporterClassifiers defines rules for exporter classification
 	ExporterClassifiers []ExporterClassifierRule
 	// InterfaceClassifiers defines rules for interface classification

inlet/core/config_test.go

@@ -0,0 +1,16 @@
+// SPDX-FileCopyrightText: 2022 Free Mobile
+// SPDX-License-Identifier: AGPL-3.0-only
+
+package core
+
+import (
+	"testing"
+
+	"akvorado/common/helpers"
+)
+
+func TestDefaultConfiguration(t *testing.T) {
+	if err := helpers.Validate.Struct(DefaultConfiguration()); err != nil {
+		t.Fatalf("validate.Struct() error:\n%+v", err)
+	}
+}

inlet/flow/input/file/config.go

@@ -8,7 +8,7 @@ import "akvorado/inlet/flow/input"
 // Configuration describes file input configuration.
 type Configuration struct {
 	// Paths to use as input
-	Paths []string
+	Paths []string `validate:"min=1,dive,required"`
 }
 
 // DefaultConfiguration descrives the default configuration for file input.

inlet/flow/input/file/config_test.go

@@ -0,0 +1,18 @@
+// SPDX-FileCopyrightText: 2022 Free Mobile
+// SPDX-License-Identifier: AGPL-3.0-only
+
+package file
+
+import (
+	"testing"
+
+	"akvorado/common/helpers"
+)
+
+func TestDefaultConfiguration(t *testing.T) {
+	if err := helpers.Validate.Struct(Configuration{
+		Paths: []string{"/path/1", "/path/2"},
+	}); err != nil {
+		t.Fatalf("validate.Struct() error:\n%+v", err)
+	}
+}

inlet/flow/input/udp/config.go

@@ -8,9 +8,9 @@ import "akvorado/inlet/flow/input"
 // Configuration describes UDP input configuration.
 type Configuration struct {
 	// Listen tells which port to listen to.
-	Listen string
+	Listen string `validate:"listen"`
 	// Workers define the number of workers to use for receiving flows.
-	Workers int
+	Workers int `validate:"min=1"`
 	// QueueSize defines the size of the channel used to
 	// communicate incoming flows. 0 can be used to disable
 	// buffering.

inlet/flow/input/udp/config_test.go

@@ -0,0 +1,16 @@
+// SPDX-FileCopyrightText: 2022 Free Mobile
+// SPDX-License-Identifier: AGPL-3.0-only
+
+package udp
+
+import (
+	"testing"
+
+	"akvorado/common/helpers"
+)
+
+func TestDefaultConfiguration(t *testing.T) {
+	if err := helpers.Validate.Struct(DefaultConfiguration()); err != nil {
+		t.Fatalf("validate.Struct() error:\n%+v", err)
+	}
+}

inlet/kafka/config.go

@@ -16,9 +16,9 @@ import (
 type Configuration struct {
 	kafka.Configuration `mapstructure:",squash" yaml:"-,inline"`
 	// FlushInterval tells how often to flush pending data to Kafka.
-	FlushInterval time.Duration
+	FlushInterval time.Duration `validate:"min=1s"`
 	// FlushBytes tells to flush when there are many bytes to write
-	FlushBytes int
+	FlushBytes int `validate:"min=1000"`
 	// MaxMessageBytes is the maximum permitted size of a message.
 	// Should be set equal or smaller than broker's
 	// `message.max.bytes`.
@@ -26,7 +26,7 @@ type Configuration struct {
 	// CompressionCodec defines the compression to use.
 	CompressionCodec CompressionCodec
 	// QueueSize defines the size of the channel used to send to Kafka.
-	QueueSize int
+	QueueSize int `validate:"min=0"`
 }
 
 // DefaultConfiguration represents the default configuration for the Kafka exporter.

inlet/kafka/config_test.go

@@ -6,6 +6,8 @@ package kafka
 import (
 	"testing"
 
+	"akvorado/common/helpers"
+
 	"github.com/Shopify/sarama"
 )
 
@@ -37,3 +39,9 @@ func TestCompressionCodecUnmarshal(t *testing.T) {
 		}
 	}
 }
+
+func TestDefaultConfiguration(t *testing.T) {
+	if err := helpers.Validate.Struct(DefaultConfiguration()); err != nil {
+		t.Fatalf("validate.Struct() error:\n%+v", err)
+	}
+}

inlet/snmp/config.go

@@ -10,25 +10,25 @@ import (
 // Configuration describes the configuration for the SNMP client
 type Configuration struct {
 	// CacheDuration defines how long to keep cached entries without access
-	CacheDuration time.Duration
+	CacheDuration time.Duration `validate:"min=1m"`
 	// CacheRefresh defines how soon to refresh an existing cached entry
-	CacheRefresh time.Duration
+	CacheRefresh time.Duration `validate:"eq=0|min=1m,eq=0|gtefield=CacheDuration"`
 	// CacheRefreshInterval defines the interval to check for expiration/refresh
-	CacheCheckInterval time.Duration
+	CacheCheckInterval time.Duration `validate:gtefield=CacheRefresh"`
 	// CachePersist defines a file to store cache and survive restarts
 	CachePersistFile string
 	// DefaultCommunity is the default SNMP community to use
-	DefaultCommunity string
+	DefaultCommunity string `validate:"required"`
 	// Communities is a mapping from exporter IPs to communities
 	Communities map[string]string
 	// PollerRetries tell how many time a poller should retry before giving up
-	PollerRetries int
+	PollerRetries int `validate:"min=0"`
 	// PollerTimeout tell how much time a poller should wait for an answer
 	PollerTimeout time.Duration
 	// PollerCoalesce tells how many requests can be contained inside a single SNMP PDU
-	PollerCoalesce int
+	PollerCoalesce int `validate:"min=0"`
 	// Workers define the number of workers used to poll SNMP
-	Workers int
+	Workers int `validate:"min=1"`
 }
 
 // DefaultConfiguration represents the default configuration for the SNMP client.

inlet/snmp/config_test.go

@@ -0,0 +1,16 @@
+// SPDX-FileCopyrightText: 2022 Free Mobile
+// SPDX-License-Identifier: AGPL-3.0-only
+
+package snmp
+
+import (
+	"testing"
+
+	"akvorado/common/helpers"
+)
+
+func TestDefaultConfiguration(t *testing.T) {
+	if err := helpers.Validate.Struct(DefaultConfiguration()); err != nil {
+		t.Fatalf("validate.Struct() error:\n%+v", err)
+	}
+}

orchestrator/clickhouse/config.go

@@ -32,7 +32,7 @@ type Configuration struct {
 	Networks NetworkNames
 	// OrchestratorURL allows one to override URL to reach
 	// orchestrator from Clickhouse
-	OrchestratorURL string
+	OrchestratorURL string `validate:"isdefault|url"`
 }
 
 // ResolutionConfiguration describes a consolidation interval.
@@ -50,7 +50,7 @@ type ResolutionConfiguration struct {
 type KafkaConfiguration struct {
 	kafka.Configuration `mapstructure:",squash" yaml:"-,inline"`
 	// Consumers tell how many consumers to use to poll data from Kafka
-	Consumers int
+	Consumers int `validate:"min=1"`
 }
 
 // DefaultConfiguration represents the default configuration for the ClickHouse configurator.

orchestrator/clickhouse/config_test.go

@@ -64,3 +64,11 @@ func TestNetworkNamesUnmarshalHook(t *testing.T) {
 		})
 	}
 }
+
+func TestDefaultConfiguration(t *testing.T) {
+	config := DefaultConfiguration()
+	config.Kafka.Topic = "flow"
+	if err := helpers.Validate.Struct(config); err != nil {
+		t.Fatalf("validate.Struct() error:\n%+v", err)
+	}
+}

orchestrator/kafka/config.go

@@ -15,9 +15,9 @@ type Configuration struct {
 // TopicConfiguration describes the configuration for a topic
 type TopicConfiguration struct {
 	// NumPartitions tells how many partitions should be used for the topic.
-	NumPartitions int32
+	NumPartitions int32 `validate:"min=1"`
 	// ReplicationFactor tells the replication factor for the topic.
-	ReplicationFactor int16
+	ReplicationFactor int16 `validate:"min=1"`
 	// ConfigEntries is a map to specify the topic overrides. Non-listed overrides will be removed
 	ConfigEntries map[string]*string
 }

orchestrator/kafka/config_test.go

@@ -0,0 +1,16 @@
+// SPDX-FileCopyrightText: 2022 Free Mobile
+// SPDX-License-Identifier: AGPL-3.0-only
+
+package kafka
+
+import (
+	"testing"
+
+	"akvorado/common/helpers"
+)
+
+func TestDefaultConfiguration(t *testing.T) {
+	if err := helpers.Validate.Struct(DefaultConfiguration()); err != nil {
+		t.Fatalf("validate.Struct() error:\n%+v", err)
+	}
+}