Adding handler functions and tests (#65)

* Adding handler functions and tests

* addressing comments

* s/service/namespace/

handlers/namespaces.go

@@ -0,0 +1,88 @@
+// Copyright (c) 2017 Uber Technologies, Inc.
+//
+// Permission is hereby granted, free of charge, to any person obtaining a copy
+// of this software and associated documentation files (the "Software"), to deal
+// in the Software without restriction, including without limitation the rights
+// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+// copies of the Software, and to permit persons to whom the Software is
+// furnished to do so, subject to the following conditions:
+//
+// The above copyright notice and this permission notice shall be included in
+// all copies or substantial portions of the Software.
+//
+// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+// THE SOFTWARE.
+
+package handlers
+
+import (
+	"fmt"
+
+	"github.com/m3db/m3cluster/kv"
+	"github.com/m3db/m3metrics/generated/proto/schema"
+)
+
+// Namespaces returns the version and the persisted namespaces in kv store.
+func Namespaces(store kv.Store, namespacesKey string) (int, *schema.Namespaces, error) {
+	value, err := store.Get(namespacesKey)
+	if err != nil {
+		return 0, nil, err
+	}
+
+	version := value.Version()
+	var namespaces schema.Namespaces
+	if err := value.Unmarshal(&namespaces); err != nil {
+		return 0, nil, err
+	}
+
+	return version, &namespaces, nil
+}
+
+// ValidateNamespace validates whether a given namespace exists.
+func ValidateNamespace(store kv.Store, namespacesKey string, namespaceName string) (int, *schema.Namespaces, *schema.Namespace, error) {
+	namespacesVersion, namespaces, err := Namespaces(store, namespacesKey)
+	if err != nil {
+		return 0, nil, nil, fmt.Errorf("could not read namespaces data: %v", err)
+	}
+	ns, err := Namespace(namespaces, namespaceName)
+	if err != nil {
+		return 0, nil, nil, fmt.Errorf("error finding namespace with name %s: %v", namespaceName, err)
+	}
+	if ns == nil {
+		return 0, nil, nil, fmt.Errorf("namespace %s doesn't exist", namespaceName)
+	}
+	if len(ns.Snapshots) == 0 {
+		return 0, nil, nil, fmt.Errorf("namespace %s has no snapshots", namespaceName)
+	}
+	if ns.Snapshots[len(ns.Snapshots)-1].Tombstoned {
+		return 0, nil, nil, fmt.Errorf("namespace %s is tombstoned", namespaceName)
+	}
+	return namespacesVersion, namespaces, ns, nil
+}
+
+// Namespace returns the namespace with a given name, or an error if there are
+// multiple matches.
+func Namespace(namespaces *schema.Namespaces, namespaceName string) (*schema.Namespace, error) {
+	var namespace *schema.Namespace
+	for _, ns := range namespaces.Namespaces {
+		if ns.Name != namespaceName {
+			continue
+		}
+		if namespace == nil {
+			namespace = ns
+		} else {
+			return nil, errMultipleMatches
+		}
+	}
+
+	if namespace == nil {
+		return nil, kv.ErrNotFound
+	}
+
+	return namespace, nil
+}

handlers/namespaces_test.go

@@ -0,0 +1,134 @@
+// Copyright (c) 2017 Uber Technologies, Inc.
+//
+// Permission is hereby granted, free of charge, to any person obtaining a copy
+// of this software and associated documentation files (the "Software"), to deal
+// in the Software without restriction, including without limitation the rights
+// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+// copies of the Software, and to permit persons to whom the Software is
+// furnished to do so, subject to the following conditions:
+//
+// The above copyright notice and this permission notice shall be included in
+// all copies or substantial portions of the Software.
+//
+// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+// THE SOFTWARE.
+
+package handlers
+
+import (
+	"testing"
+
+	"github.com/m3db/m3cluster/kv"
+	"github.com/m3db/m3cluster/kv/mem"
+	"github.com/m3db/m3metrics/generated/proto/schema"
+	"github.com/stretchr/testify/require"
+)
+
+const (
+	testNamespaceKey = "testKey"
+)
+
+var (
+	testNamespaces = &schema.Namespaces{
+		Namespaces: []*schema.Namespace{
+			&schema.Namespace{
+				Name: "fooNs",
+				Snapshots: []*schema.NamespaceSnapshot{
+					&schema.NamespaceSnapshot{
+						ForRulesetVersion: 1,
+						Tombstoned:        false,
+					},
+					&schema.NamespaceSnapshot{
+						ForRulesetVersion: 2,
+						Tombstoned:        false,
+					},
+				},
+			},
+			&schema.Namespace{
+				Name: "barNs",
+				Snapshots: []*schema.NamespaceSnapshot{
+					&schema.NamespaceSnapshot{
+						ForRulesetVersion: 1,
+						Tombstoned:        false,
+					},
+					&schema.NamespaceSnapshot{
+						ForRulesetVersion: 2,
+						Tombstoned:        true,
+					},
+				},
+			},
+		},
+	}
+
+	badNamespaces = &schema.Namespaces{
+		Namespaces: []*schema.Namespace{
+			&schema.Namespace{Name: "fooNs", Snapshots: nil},
+			&schema.Namespace{Name: "fooNs", Snapshots: nil},
+		},
+	}
+)
+
+func TestNamespace(t *testing.T) {
+	res, err := Namespace(testNamespaces, "barNs")
+	require.NoError(t, err)
+	require.EqualValues(t, testNamespaces.Namespaces[1], res)
+}
+
+func TestNamespaceError(t *testing.T) {
+	res, err := Namespace(badNamespaces, "blah")
+	require.Error(t, err)
+	require.Equal(t, err, kv.ErrNotFound)
+	require.Nil(t, res)
+
+	res, err = Namespace(badNamespaces, "fooNs")
+	require.Error(t, err)
+	require.Equal(t, err, errMultipleMatches)
+	require.Nil(t, res)
+}
+
+func TestNamespaces(t *testing.T) {
+	store := mem.NewStore()
+	store.Set(testNamespaceKey, testNamespaces)
+	_, s, err := Namespaces(store, testNamespaceKey)
+	require.NoError(t, err)
+	require.NotNil(t, s.Namespaces)
+}
+
+func TestNamespacesError(t *testing.T) {
+	store := mem.NewStore()
+	store.Set(testNamespaceKey, &schema.RollupRule{Uuid: "x"})
+	_, s, err := Namespaces(store, testNamespaceKey)
+	require.Error(t, err)
+	require.Nil(t, s)
+}
+
+func TestValidateNamespace(t *testing.T) {
+	store := mem.NewStore()
+	store.Set(testNamespaceKey, testNamespaces)
+	v, s, ns, err := ValidateNamespace(store, testNamespaceKey, "fooNs")
+	require.Equal(t, v, 1)
+	require.NoError(t, err)
+	require.NotNil(t, s.Namespaces, nil)
+	require.Equal(t, ns.Name, "fooNs")
+}
+
+func TestValidateNamespaceDNE(t *testing.T) {
+	store := mem.NewStore()
+	store.Set(testNamespaceKey, testNamespaces)
+	_, _, _, err := ValidateNamespace(store, testNamespaceKey, "blah")
+	require.Error(t, err)
+	require.Contains(t, err.Error(), "not found")
+}
+
+func TestValidateNamespaceTombstoned(t *testing.T) {
+	store := mem.NewStore()
+	store.Set(testNamespaceKey, testNamespaces)
+	_, _, _, err := ValidateNamespace(store, testNamespaceKey, "barNs")
+	require.Error(t, err)
+	require.Contains(t, err.Error(), "tombstoned")
+}

handlers/ruleset.go

@@ -0,0 +1,112 @@
+// Copyright (c) 2017 Uber Technologies, Inc.
+//
+// Permission is hereby granted, free of charge, to any person obtaining a copy
+// of this software and associated documentation files (the "Software"), to deal
+// in the Software without restriction, including without limitation the rights
+// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+// copies of the Software, and to permit persons to whom the Software is
+// furnished to do so, subject to the following conditions:
+//
+// The above copyright notice and this permission notice shall be included in
+// all copies or substantial portions of the Software.
+//
+// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+// THE SOFTWARE.
+
+package handlers
+
+import (
+	"errors"
+	"fmt"
+
+	"github.com/m3db/m3cluster/kv"
+	"github.com/m3db/m3metrics/generated/proto/schema"
+)
+
+var (
+	errMultipleMatches = errors.New("more than one match found")
+)
+
+// Rule returns the rule with a given name, or an error if there are mutliple matches
+func Rule(ruleSet *schema.RuleSet, ruleName string) (*schema.MappingRule, *schema.RollupRule, error) {
+	var (
+		mappingRule *schema.MappingRule
+		rollupRule  *schema.RollupRule
+	)
+	for _, mr := range ruleSet.MappingRules {
+		if len(mr.Snapshots) == 0 {
+			continue
+		}
+
+		latestSnapshot := mr.Snapshots[len(mr.Snapshots)-1]
+		name := latestSnapshot.Name
+		if name != ruleName || latestSnapshot.Tombstoned {
+			continue
+		}
+		if mappingRule == nil {
+			mappingRule = mr
+		} else {
+			return nil, nil, errMultipleMatches
+		}
+	}
+	for _, rr := range ruleSet.RollupRules {
+		if len(rr.Snapshots) == 0 {
+			continue
+		}
+		latestSnapshot := rr.Snapshots[len(rr.Snapshots)-1]
+		name := latestSnapshot.Name
+		if name != ruleName || latestSnapshot.Tombstoned {
+			continue
+		}
+		if rollupRule == nil {
+			rollupRule = rr
+		} else {
+			return nil, nil, errMultipleMatches
+		}
+	}
+	if mappingRule != nil && rollupRule != nil {
+		return nil, nil, errMultipleMatches
+	}
+
+	if mappingRule == nil && rollupRule == nil {
+		return nil, nil, kv.ErrNotFound
+	}
+	return mappingRule, rollupRule, nil
+}
+
+// RuleSet returns the version and the persisted ruleset data in kv store.
+func RuleSet(store kv.Store, ruleSetKey string) (int, *schema.RuleSet, error) {
+	value, err := store.Get(ruleSetKey)
+	if err != nil {
+		return 0, nil, err
+	}
+	version := value.Version()
+	var ruleSet schema.RuleSet
+	if err := value.Unmarshal(&ruleSet); err != nil {
+		return 0, nil, err
+	}
+
+	return version, &ruleSet, nil
+}
+
+// RuleSetKey returns the ruleset key given the namespace name.
+func RuleSetKey(keyFmt string, namespace string) string {
+	return fmt.Sprintf(keyFmt, namespace)
+}
+
+// ValidateRuleSet validates that a valid RuleSet exists in that keyspace.
+func ValidateRuleSet(store kv.Store, ruleSetKey string) (int, *schema.RuleSet, error) {
+	ruleSetVersion, ruleSet, err := RuleSet(store, ruleSetKey)
+	if err != nil {
+		return 0, nil, fmt.Errorf("could not read ruleSet data for key %s: %v", ruleSetKey, err)
+	}
+	if ruleSet.Tombstoned {
+		return 0, nil, fmt.Errorf("ruleset %s is tombstoned", ruleSetKey)
+	}
+	return ruleSetVersion, ruleSet, nil
+}