common/schema: turns into a component

This is a first step to make it accept configuration. Most of the
changes are quite trivial, but I also ran into some difficulties with
query columns and filters. They need the schema for parsing, but parsing
happens before dependencies are instantiated (and even if it was not the
case, parsing is stateless). Therefore, I have added a `Validate()`
method that must be called after instantiation. Various bits `panic()`
if not validated to ensure we catch all cases.

The alternative to make the component manages a global state would have
been simpler but it would break once we add the ability to add or
disable columns.

cmd/console.go

@@ -12,6 +12,7 @@ import (
 	"akvorado/common/daemon"
 	"akvorado/common/http"
 	"akvorado/common/reporter"
+	"akvorado/common/schema"
 	"akvorado/console"
 	"akvorado/console/authentication"
 	"akvorado/console/database"
@@ -103,12 +104,17 @@ func consoleStart(r *reporter.Reporter, config ConsoleConfiguration, checkOnly b
 	if err != nil {
 		return fmt.Errorf("unable to initialize database component: %w", err)
 	}
+	schemaComponent, err := schema.New()
+	if err != nil {
+		return fmt.Errorf("unable to initialize schema component: %w", err)
+	}
 	consoleComponent, err := console.New(r, config.Console, console.Dependencies{
 		Daemon:       daemonComponent,
 		HTTP:         httpComponent,
 		ClickHouseDB: clickhouseComponent,
 		Auth:         authenticationComponent,
 		Database:     databaseComponent,
+		Schema:       schemaComponent,
 	})
 	if err != nil {
 		return fmt.Errorf("unable to initialize console component: %w", err)

cmd/inlet.go

@@ -11,6 +11,7 @@ import (
 	"akvorado/common/daemon"
 	"akvorado/common/http"
 	"akvorado/common/reporter"
+	"akvorado/common/schema"
 	"akvorado/inlet/bmp"
 	"akvorado/inlet/core"
 	"akvorado/inlet/flow"
@@ -95,9 +96,14 @@ func inletStart(r *reporter.Reporter, config InletConfiguration, checkOnly bool)
 	if err != nil {
 		return fmt.Errorf("unable to initialize http component: %w", err)
 	}
+	schemaComponent, err := schema.New()
+	if err != nil {
+		return fmt.Errorf("unable to initialize schema component: %w", err)
+	}
 	flowComponent, err := flow.New(r, config.Flow, flow.Dependencies{
 		Daemon: daemonComponent,
 		HTTP:   httpComponent,
+		Schema: schemaComponent,
 	})
 	if err != nil {
 		return fmt.Errorf("unable to initialize flow component: %w", err)
@@ -122,6 +128,7 @@ func inletStart(r *reporter.Reporter, config InletConfiguration, checkOnly bool)
 	}
 	kafkaComponent, err := kafka.New(r, config.Kafka, kafka.Dependencies{
 		Daemon: daemonComponent,
+		Schema: schemaComponent,
 	})
 	if err != nil {
 		return fmt.Errorf("unable to initialize Kafka component: %w", err)
@@ -134,6 +141,7 @@ func inletStart(r *reporter.Reporter, config InletConfiguration, checkOnly bool)
 		GeoIP:  geoipComponent,
 		Kafka:  kafkaComponent,
 		HTTP:   httpComponent,
+		Schema: schemaComponent,
 	})
 	if err != nil {
 		return fmt.Errorf("unable to initialize core component: %w", err)

cmd/orchestrator.go

@@ -12,6 +12,7 @@ import (
 	"akvorado/common/daemon"
 	"akvorado/common/http"
 	"akvorado/common/reporter"
+	"akvorado/common/schema"
 	"akvorado/orchestrator"
 	"akvorado/orchestrator/clickhouse"
 	"akvorado/orchestrator/kafka"
@@ -111,7 +112,11 @@ func orchestratorStart(r *reporter.Reporter, config OrchestratorConfiguration, c
 	if err != nil {
 		return fmt.Errorf("unable to initialize HTTP component: %w", err)
 	}
-	kafkaComponent, err := kafka.New(r, config.Kafka)
+	schemaComponent, err := schema.New()
+	if err != nil {
+		return fmt.Errorf("unable to initialize schema component: %w", err)
+	}
+	kafkaComponent, err := kafka.New(r, config.Kafka, kafka.Dependencies{Schema: schemaComponent})
 	if err != nil {
 		return fmt.Errorf("unable to initialize kafka component: %w", err)
 	}
@@ -125,6 +130,7 @@ func orchestratorStart(r *reporter.Reporter, config OrchestratorConfiguration, c
 		Daemon:     daemonComponent,
 		HTTP:       httpComponent,
 		ClickHouse: clickhouseDBComponent,
+		Schema:     schemaComponent,
 	})
 	if err != nil {
 		return fmt.Errorf("unable to initialize clickhouse component: %w", err)

common/helpers/tests_diff.go

@@ -23,12 +23,24 @@ var prettyC = pretty.Config{
 }
 
 func defaultPrettyFormatters() map[reflect.Type]interface{} {
-	return map[reflect.Type]interface{}{
+	result := map[reflect.Type]interface{}{
 		reflect.TypeOf(net.IP{}):            fmt.Sprint,
 		reflect.TypeOf(netip.Addr{}):        fmt.Sprint,
 		reflect.TypeOf(time.Time{}):         fmt.Sprint,
 		reflect.TypeOf(SubnetMap[string]{}): fmt.Sprint,
 	}
+	for t, fn := range nonDefaultPrettyFormatters {
+		result[t] = fn
+	}
+	return result
+}
+
+var nonDefaultPrettyFormatters = map[reflect.Type]interface{}{}
+
+// AddPrettyFormatter add a global pretty formatter. We cannot put everything in
+// the default map due to cycles.
+func AddPrettyFormatter(t reflect.Type, fn interface{}) {
+	nonDefaultPrettyFormatters[t] = fn
 }
 
 // DiffOption changes the behavior of the Diff function.