Add namespace validation logic (#125)

rules/store.go

@@ -22,15 +22,18 @@ package rules
 
 // Store performs read/write operations for rules and namespaces.
 type Store interface {
+	// ReadNamespaces returns the persisted namespaces in kv store.
+	ReadNamespaces() (*Namespaces, error)
+
+	// ReadRuleSet returns the persisted ruleset in kv store.
+	ReadRuleSet(nsName string) (RuleSet, error)
+
 	// WriteRuleSet saves the given ruleset to the backing store.
 	WriteRuleSet(rs MutableRuleSet) error
 
 	// WriteAll saves both the given ruleset and namespace to the backing store.
 	WriteAll(nss *Namespaces, rs MutableRuleSet) error
 
-	// ReadNamespaces returns the persisted namespaces in kv store.
-	ReadNamespaces() (*Namespaces, error)
-
-	// ReadRuleSet returns the persisted ruleset in kv store.
-	ReadRuleSet(nsName string) (RuleSet, error)
+	// Close closes the store.
+	Close()
 }

rules/store/kv/store.go

@@ -127,6 +127,13 @@ func (s *store) WriteAll(nss *rules.Namespaces, rs rules.MutableRuleSet) error {
 	return err
 }
 
+func (s *store) Close() {
+	if s.opts.Validator == nil {
+		return
+	}
+	s.opts.Validator.Close()
+}
+
 func (s *store) ruleSetKey(ns string) string {
 	return fmt.Sprintf(s.opts.RuleSetKeyFmt, ns)
 }

rules/store/kv/store_test.go

@@ -21,6 +21,7 @@
 package kv
 
 import (
+	"errors"
 	"fmt"
 	"testing"
 	"time"
@@ -339,6 +340,8 @@ var (
 
 func TestRuleSetKey(t *testing.T) {
 	s := testStore()
+	defer s.Close()
+
 	key := s.(*store).ruleSetKey(testNamespace)
 	require.Equal(t, "rules/fooNs", key)
 }
@@ -347,22 +350,27 @@ func TestNewStore(t *testing.T) {
 	opts := NewStoreOptions(testNamespaceKey, testRuleSetKeyFmt, nil)
 	kvStore := mem.NewStore()
 	s := NewStore(kvStore, opts).(*store)
+	defer s.Close()
 
 	require.Equal(t, s.kvStore, kvStore)
 	require.Equal(t, s.opts, opts)
 }
 
 func TestReadNamespaces(t *testing.T) {
 	s := testStore()
+	defer s.Close()
+
 	_, e := s.(*store).kvStore.Set(testNamespaceKey, testNamespaces)
 	require.NoError(t, e)
 	nss, err := s.ReadNamespaces()
 	require.NoError(t, err)
 	require.NotNil(t, nss.Namespaces)
 }
 
-func TestNamespacesError(t *testing.T) {
+func TestReadNamespacesError(t *testing.T) {
 	s := testStore()
+	defer s.Close()
+
 	_, e := s.(*store).kvStore.Set(testNamespaceKey, &schema.RollupRule{Uuid: "x"})
 	require.NoError(t, e)
 	nss, err := s.ReadNamespaces()
@@ -372,24 +380,29 @@ func TestNamespacesError(t *testing.T) {
 
 func TestReadRuleSet(t *testing.T) {
 	s := testStore()
+	defer s.Close()
+
 	_, e := s.(*store).kvStore.Set(testRuleSetKey, testRuleSet)
 	require.NoError(t, e)
 	rs, err := s.ReadRuleSet(testNamespace)
 	require.NoError(t, err)
 	require.NotNil(t, rs)
 }
 
-func TestRuleSetError(t *testing.T) {
+func TestReadRuleSetError(t *testing.T) {
 	s := testStore()
+	defer s.Close()
+
 	_, e := s.(*store).kvStore.Set(testRuleSetKey, &schema.Namespace{Name: "x"})
 	require.NoError(t, e)
 	rs, err := s.ReadRuleSet("blah")
 	require.Error(t, err)
 	require.Nil(t, rs)
 }
 
-func TestWrite(t *testing.T) {
+func TestWriteAll(t *testing.T) {
 	s := testStore()
+	defer s.Close()
 
 	rs, err := s.ReadRuleSet(testNamespaceKey)
 	require.Error(t, err)
@@ -419,8 +432,19 @@ func TestWrite(t *testing.T) {
 	require.Equal(t, nssSchema, testNamespaces)
 }
 
-func TestWriteErrorAll(t *testing.T) {
+func TestWriteAllValidationError(t *testing.T) {
+	errInvalidRuleSet := errors.New("invalid ruleset")
+	v := &mockValidator{
+		validateFn: func(rules.RuleSet) error { return errInvalidRuleSet },
+	}
+	s := testStoreWithValidator(v)
+	defer s.Close()
+	require.Equal(t, errInvalidRuleSet, s.WriteAll(nil, nil))
+}
+
+func TestWriteAllError(t *testing.T) {
 	s := testStore()
+	defer s.Close()
 
 	rs, err := s.ReadRuleSet(testNamespaceKey)
 	require.Error(t, err)
@@ -461,8 +485,19 @@ func TestWriteErrorAll(t *testing.T) {
 	require.Error(t, err)
 }
 
-func TestWriteErrorRuleSet(t *testing.T) {
+func TestWriteRuleSetValidationError(t *testing.T) {
+	errInvalidRuleSet := errors.New("invalid ruleset")
+	v := &mockValidator{
+		validateFn: func(rules.RuleSet) error { return errInvalidRuleSet },
+	}
+	s := testStoreWithValidator(v)
+	defer s.Close()
+	require.Equal(t, errInvalidRuleSet, s.WriteRuleSet(nil))
+}
+
+func TestWriteRuleSetError(t *testing.T) {
 	s := testStore()
+	defer s.Close()
 
 	rs, err := s.ReadRuleSet(testNamespaceKey)
 	require.Error(t, err)
@@ -486,8 +521,9 @@ func TestWriteErrorRuleSet(t *testing.T) {
 	require.Error(t, err)
 }
 
-func TestWriteNoNamespace(t *testing.T) {
+func TestWriteAllNoNamespace(t *testing.T) {
 	s := testStore()
+	defer s.Close()
 
 	rs, err := s.ReadRuleSet(testNamespaceKey)
 	require.Error(t, err)
@@ -522,7 +558,11 @@ func TestWriteNoNamespace(t *testing.T) {
 }
 
 func testStore() rules.Store {
-	opts := NewStoreOptions(testNamespaceKey, testRuleSetKeyFmt, nil)
+	return testStoreWithValidator(nil)
+}
+
+func testStoreWithValidator(validator rules.Validator) rules.Store {
+	opts := NewStoreOptions(testNamespaceKey, testRuleSetKeyFmt, validator)
 	kvStore := mem.NewStore()
 	return NewStore(kvStore, opts)
 }
@@ -538,3 +578,13 @@ func newMutableRuleSetFromSchema(
 	require.NoError(t, err)
 	return roRuleSet.ToMutableRuleSet()
 }
+
+type validateFn func(rs rules.RuleSet) error
+
+type mockValidator struct {
+	validateFn validateFn
+}
+
+func (v *mockValidator) Validate(rs rules.RuleSet) error                        { return v.validateFn(rs) }
+func (v *mockValidator) ValidateSnapshot(snapshot *rules.RuleSetSnapshot) error { return nil }
+func (v *mockValidator) Close()                                                 {}

rules/validator.go

@@ -27,4 +27,7 @@ type Validator interface {
 
 	// ValidateSnapshot validates a ruleset snapshot.
 	ValidateSnapshot(snapshot *RuleSetSnapshot) error
+
+	// Close closes the validator.
+	Close()
 }

rules/validator/config.go

@@ -21,14 +21,26 @@
 package validator
 
 import (
+	"errors"
+
+	"github.com/m3db/m3cluster/client"
 	"github.com/m3db/m3metrics/filters"
 	"github.com/m3db/m3metrics/metric"
 	"github.com/m3db/m3metrics/policy"
 	"github.com/m3db/m3metrics/rules"
+	"github.com/m3db/m3metrics/rules/validator/namespace"
+	"github.com/m3db/m3metrics/rules/validator/namespace/kv"
+	"github.com/m3db/m3metrics/rules/validator/namespace/static"
+)
+
+var (
+	errNoNamespaceValidatorConfiguration        = errors.New("no namespace validator configuration provided")
+	errMultipleNamespaceValidatorConfigurations = errors.New("multiple namespace validator configurations provided")
 )
 
 // Configuration is the configuration for rules validation.
 type Configuration struct {
+	Namespace              namespaceValidatorConfiguration    `yaml:"namespace"`
 	RequiredRollupTags     []string                           `yaml:"requiredRollupTags"`
 	MetricTypes            metricTypesValidationConfiguration `yaml:"metricTypes"`
 	Policies               policiesValidationConfiguration    `yaml:"policies"`
@@ -37,8 +49,15 @@ type Configuration struct {
 }
 
 // NewValidator creates a new rules validator based on the given configuration.
-func (c Configuration) NewValidator() rules.Validator {
+func (c Configuration) NewValidator(
+	kvClient client.Client,
+) (rules.Validator, error) {
+	nsValidator, err := c.Namespace.NewNamespaceValidator(kvClient)
+	if err != nil {
+		return nil, err
+	}
 	opts := NewOptions().
+		SetNamespaceValidator(nsValidator).
 		SetRequiredRollupTags(c.RequiredRollupTags).
 		SetMetricTypesFn(c.MetricTypes.NewMetricTypesFn()).
 		SetDefaultAllowedStoragePolicies(c.Policies.DefaultAllowed.StoragePolicies).
@@ -50,7 +69,27 @@ func (c Configuration) NewValidator() rules.Validator {
 			SetAllowedStoragePoliciesFor(override.Type, override.Allowed.StoragePolicies).
 			SetAllowedCustomAggregationTypesFor(override.Type, override.Allowed.AggregationTypes)
 	}
-	return NewValidator(opts)
+	return NewValidator(opts), nil
+}
+
+type namespaceValidatorConfiguration struct {
+	KV     *kv.NamespaceValidatorConfiguration     `yaml:"kv"`
+	Static *static.NamespaceValidatorConfiguration `yaml:"static"`
+}
+
+func (c namespaceValidatorConfiguration) NewNamespaceValidator(
+	kvClient client.Client,
+) (namespace.Validator, error) {
+	if c.KV == nil && c.Static == nil {
+		return nil, errNoNamespaceValidatorConfiguration
+	}
+	if c.KV != nil && c.Static != nil {
+		return nil, errMultipleNamespaceValidatorConfigurations
+	}
+	if c.KV != nil {
+		return c.KV.NewNamespaceValidator(kvClient)
+	}
+	return c.Static.NewNamespaceValidator(), nil
 }
 
 // metricTypesValidationConfiguration is th configuration for metric types validation.

rules/validator/config_test.go

@@ -23,13 +23,79 @@ package validator
 import (
 	"testing"
 
+	"github.com/m3db/m3cluster/client"
+	"github.com/m3db/m3cluster/kv"
+	"github.com/m3db/m3cluster/kv/mem"
 	"github.com/m3db/m3metrics/filters"
 	"github.com/m3db/m3metrics/metric"
 
+	"github.com/golang/mock/gomock"
 	"github.com/stretchr/testify/require"
 	yaml "gopkg.in/yaml.v2"
 )
 
+func TestNamespaceValidatorConfigurationNoConfigurationProvided(t *testing.T) {
+	cfgStr := ""
+	var cfg namespaceValidatorConfiguration
+	require.NoError(t, yaml.Unmarshal([]byte(cfgStr), &cfg))
+	_, err := cfg.NewNamespaceValidator(nil)
+	require.Equal(t, errNoNamespaceValidatorConfiguration, err)
+}
+
+func TestNamespaceValidatorConfigurationMultipleConfigurationProvided(t *testing.T) {
+	cfgStr := `
+kv:
+  kvConfig:
+    zone: testZone
+    environment: testEnvironment
+  initWatchTimeout: 5ms
+  validNamespacesKey: testValidNamespaces
+static:
+  validationResult: valid
+`
+	var cfg namespaceValidatorConfiguration
+	require.NoError(t, yaml.Unmarshal([]byte(cfgStr), &cfg))
+	_, err := cfg.NewNamespaceValidator(nil)
+	require.Equal(t, errMultipleNamespaceValidatorConfigurations, err)
+}
+
+func TestNamespaceValidatorConfigurationKV(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	defer ctrl.Finish()
+
+	cfgStr := `
+kv:
+  kvConfig:
+    zone: testZone
+    environment: testEnvironment
+  initWatchTimeout: 5ms
+  validNamespacesKey: testValidNamespaces
+`
+	var cfg namespaceValidatorConfiguration
+	require.NoError(t, yaml.Unmarshal([]byte(cfgStr), &cfg))
+
+	kvStore := mem.NewStore()
+	kvOpts := kv.NewOptions().SetZone("testZone").SetEnvironment("testEnvironment")
+	kvClient := client.NewMockClient(ctrl)
+	kvClient.EXPECT().Store(kvOpts).Return(kvStore, nil)
+	_, err := cfg.NewNamespaceValidator(kvClient)
+	require.NoError(t, err)
+}
+
+func TestNamespaceValidatorConfigurationStatic(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	defer ctrl.Finish()
+
+	cfgStr := `
+static:
+  validationResult: valid
+`
+	var cfg namespaceValidatorConfiguration
+	require.NoError(t, yaml.Unmarshal([]byte(cfgStr), &cfg))
+	_, err := cfg.NewNamespaceValidator(nil)
+	require.NoError(t, err)
+}
+
 func TestConfigurationRequiredRollupTags(t *testing.T) {
 	cfg := `
 requiredRollupTags: