Repository: jumptrading/influx-spout (go)

Issue:
More configuration validation
There are ways that configuration could be invalid which are currently silently ignored. 

For example:
* listeners and filters only care about the first topic in the `nats_topic` configuration and should fail if more than one is provided
* for the writer's filter rules ignore the `channel` field. The writer should fail if a channel is provided as this indicates a potential misunderstanding by whomever generated the configuration.
* listener should refuse to start if `batch_messages` is 0 (or use a sensible default)

Pull request description:
Introduce configuration validation
A number of possible configuration errors are detected and more checks can be added as required.

Closes #30.

Patch:
diff --git a/config/config.go b/config/config.go
index 97afafb..08009ff 100644
--- a/config/config.go
+++ b/config/config.go
@@ -85,6 +85,81 @@ func newDefaultConfig() *Config {
 	}
 }
 
+// Validate returns an error if a problem is detected with the
+// configuration. It checks a variety of possible configuration errors
+// but these aren't exhaustive.
+func (c *Config) Validate() error {
+	if c.ProbePort < 0 || c.ProbePort > 65535 {
+		return errors.New("probe_port out of range")
+	}
+	if c.PprofPort < 0 || c.PprofPort > 65535 {
+		return errors.New("pprof_port out of range")
+	}
+
+	switch c.Mode {
+	case "listener", "listener_http":
+		return c.validateListener()
+	case "filter":
+		return c.validateFilter()
+	case "writer":
+		return c.validateWriter()
+	case "monitor":
+	default:
+		return fmt.Errorf("invalid mode: %s", c.Mode)
+	}
+	return nil
+}
+
+func (c *Config) validateListener() error {
+	if c.Port < 1 || c.Port > 65535 {
+		return errors.New("listener port out of range")
+	}
+	if len(c.NATSSubject) != 1 {
+		return errors.New("listener should only use one NATS subject")
+	}
+	return nil
+}
+
+func (c *Config) validateFilter() error {
+	if len(c.NATSSubject) != 1 {
+		return errors.New("filter should only use one NATS subject")
+	}
+	for _, rule := range c.Rule {
+		if rule.Rtype == "" {
+			return errors.New(`rule missing "type"`)
+		}
+		if rule.Match == "" {
+			return errors.New(`rule missing "match"`)
+		}
+		if rule.Subject == "" {
+			return errors.New(`rule missing "subject"`)
+		}
+	}
+	return nil
+}
+
+func (c *Config) validateWriter() error {
+	if len(c.NATSSubject) < 1 {
+		return errors.New("writer needs at least one NATS subject")
+	}
+	if c.InfluxDBPort < 1 || c.InfluxDBPort > 65535 {
+		return errors.New("influxdb_port out of range")
+	}
+
+	for _, rule := range c.Rule {
+		if rule.Rtype == "" {
+			return errors.New(`rule missing "type"`)
+		}
+		if rule.Match == "" {
+			return errors.New(`rule missing "match"`)
+		}
+		if rule.Subject != "" {
+			return fmt.Errorf("writer rules shouldn't have subject (%q)", rule.Subject)
+		}
+	}
+	return nil
+}
+
 // NewConfigFromFile parses the specified configuration file and
 // returns a Config.
 func NewConfigFromFile(fileName string) (*Config, error) {
@@ -125,6 +200,10 @@ func NewConfigFromFile(fileName string) (*Config, error) {
 		}
 	}
 
+	if err := conf.Validate(); err != nil {
+		return nil, err
+	}
+
 	return conf, nil
 }
 


Test patch:
diff --git a/config/config_small_test.go b/config/config_small_test.go
index b12b5c3..e186030 100644
--- a/config/config_small_test.go
+++ b/config/config_small_test.go
@@ -266,3 +266,169 @@ func parseConfig(content string) (*Config, error) {
 
 	return NewConfigFromFile(testConfigFileName)
 }
+
+func TestValidateDefaults(t *testing.T) {
+	conf := newDefaultConfig()
+	conf.Mode = "writer"
+	assert.NoError(t, conf.Validate())
+}
+
+func TestValidateProblems(t *testing.T) {
+	Fs = afero.NewMemMapFs()
+
+	tests := []struct {
+		config string
+		err    string
+	}{{
+		`
+mode = "frog"
+`,
+		"invalid mode: frog",
+	}, {
+		`
+mode = "listener"
+port = -1
+`,
+		"listener port out of range",
+	}, {
+		`
+mode = "listener"
+port = 65536
+`,
+		"listener port out of range",
+	}, {
+		`
+mode = "listener_http"
+port = -1
+`,
+		"listener port out of range",
+	}, {
+		`
+mode = "listener_http"
+port = 65536
+`,
+		"listener port out of range",
+	}, {
+		`
+mode = "filter"
+probe_port = 65536
+`,
+		"probe_port out of range",
+	}, {
+		`
+mode = "writer"
+probe_port = -1
+`,
+		"probe_port out of range",
+	}, {
+		`
+mode = "filter"
+pprof_port = 65536
+`,
+		"pprof_port out of range",
+	}, {
+		`
+mode = "writer"
+pprof_port = -1
+`,
+		"pprof_port out of range",
+	}, {
+		`
+mode = "listener"
+nats_subject = ["one", "two"]
+`,
+		"listener should only use one NATS subject",
+	}, {
+		`
+mode = "listener_http"
+nats_subject = ["one", "two"]
+`,
+		"listener should only use one NATS subject",
+	}, {
+		`
+mode = "filter"
+nats_subject = ["one", "two"]
+`,
+		"filter should only use one NATS subject",
+	}, {
+		`
+mode = "filter"
+
+[[rule]]
+match = "needle"
+subject = "foo"
+`,
+		`rule missing "type"`,
+	}, {
+		`
+mode = "filter"
+
+[[rule]]
+type = "basic"
+subject = "foo"
+`,
+		`rule missing "match"`,
+	}, {
+		`
+mode = "filter"
+
+[[rule]]
+type = "basic"
+match = "needle"
+`,
+		`rule missing "subject"`,
+	}, {
+		`
+mode = "writer"
+nats_subject = []
+`,
+		"writer needs at least one NATS subject",
+	}, {
+		`
+mode = "writer"
+influxdb_port = 65536
+`,
+		"influxdb_port out of range",
+	}, {
+		`
+mode = "writer"
+influxdb_port = 0
+`,
+		"influxdb_port out of range",
+	}, {
+		`
+mode = "writer"
+
+[[rule]]
+match = "needle"
+`,
+		`rule missing "type"`,
+	}, {
+		`
+mode = "writer"
+
+[[rule]]
+type = "basic"
+`,
+		`rule missing "match"`,
+	}, {
+		`
+mode = "writer"
+
+[[rule]]
+type = "basic"
+match = "needle"
+subject = "matches"
+`,
+		`writer rules shouldn't have subject ("matches")`,
+	},
+	}
+
+	for i, test := range tests {
+		t.Logf("Validate test %d (%s)", i, test.err)
+		afero.WriteFile(Fs, testConfigFileName, []byte(test.config), 0600)
+		conf, err := NewConfigFromFile(testConfigFileName)
+		assert.Nil(t, conf)
+		assert.EqualError(t, err, test.err)
+	}
+}