config: validate databroker settings (#1260)

* config: validate databroker settings

* fix test

internal/databroker/config_source.go

@@ -74,12 +74,6 @@ func (src *ConfigSource) rebuild(firstTime bool) {
 	defer src.mu.Unlock()
 
 	cfg := src.underlyingConfig.Clone()
-	defer func() {
-		src.computedConfig = cfg
-		if !firstTime {
-			src.Trigger(cfg)
-		}
-	}()
 
 	// start the updater
 	src.runUpdater(cfg)
@@ -105,6 +99,7 @@ func (src *ConfigSource) rebuild(firstTime bool) {
 				log.Warn().Err(err).
 					Str("policy", policy.String()).
 					Msg("databroker: invalid policy, ignoring")
+				continue
 			}
 
 			routeID := policy.RouteID()
@@ -119,6 +114,17 @@ func (src *ConfigSource) rebuild(firstTime bool) {
 
 			cfg.Options.Policies = append(cfg.Options.Policies, *policy)
 		}
+
+		err := cfg.Options.Validate()
+		if err != nil {
+			log.Warn().Err(err).Msg("databroker: invalid config detected, ignoring")
+			return
+		}
+	}
+
+	src.computedConfig = cfg
+	if !firstTime {
+		src.Trigger(cfg)
 	}
 }
 

internal/databroker/config_source_test.go

@@ -33,11 +33,13 @@ func TestConfigSource(t *testing.T) {
 
 	cfgs := make(chan *config.Config, 10)
 
+	base := config.NewDefaultOptions()
+	base.DataBrokerURL = mustParse("http://" + li.Addr().String())
+	base.InsecureServer = true
+	base.GRPCInsecure = true
+
 	src := NewConfigSource(config.NewStaticSource(&config.Config{
-		Options: &config.Options{
+		Options: base,
-			DataBrokerURL: mustParse("http://" + li.Addr().String()),
-			GRPCInsecure:  true,
-		},
 	}), func(cfg *config.Config) {
 		cfgs <- cfg
 	})