Make default namespace just 'default' and fix drop policy string valu…

…es (#197)

- Make default namespace just 'default' and fix drop policy string values
- Add some helper methods to strip drop staged metadatas if required

matcher/options.go

@@ -41,7 +41,7 @@ const (
 
 var (
 	defaultNamespaceTag     = []byte("namespace")
-	defaultDefaultNamespace = []byte("defaultNamespace")
+	defaultDefaultNamespace = []byte("default")
 )
 
 // RuleSetKeyFn generates the ruleset key for a given namespace.

metadata/metadata.go

@@ -258,6 +258,11 @@ func (m Metadata) IsDropPolicyApplied() bool {
 	return len(m.Pipelines) == 1 && m.Pipelines[0].IsDropPolicyApplied()
 }
 
+// Equal returns true if two metadatas are considered equal.
+func (m Metadata) Equal(other Metadata) bool {
+	return m.Pipelines.Equal(other.Pipelines)
+}
+
 // ToProto converts the metadata to a protobuf message in place.
 func (m Metadata) ToProto(pb *metricpb.Metadata) error {
 	numPipelines := len(m.Pipelines)
@@ -351,6 +356,13 @@ type StagedMetadata struct {
 	Tombstoned bool `json:"tombstoned"`
 }
 
+// Equal returns true if two staged metadatas are considered equal.
+func (sm StagedMetadata) Equal(other StagedMetadata) bool {
+	return sm.Metadata.Equal(other.Metadata) &&
+		sm.CutoverNanos == other.CutoverNanos &&
+		sm.Tombstoned == other.Tombstoned
+}
+
 // IsDefault returns whether this is a default staged metadata.
 func (sm StagedMetadata) IsDefault() bool {
 	return sm.CutoverNanos == 0 && !sm.Tombstoned && sm.Metadata.IsDefault()
@@ -385,6 +397,19 @@ func (sm *StagedMetadata) FromProto(pb metricpb.StagedMetadata) error {
 // StagedMetadatas contains a list of staged metadatas.
 type StagedMetadatas []StagedMetadata
 
+// Equal returns true if two staged metadatas slices are considered equal.
+func (sms StagedMetadatas) Equal(other StagedMetadatas) bool {
+	if len(sms) != len(other) {
+		return false
+	}
+	for i := range sms {
+		if !sms[i].Equal(other[i]) {
+			return false
+		}
+	}
+	return true
+}
+
 // IsDefault determines whether the list of staged metadata is a default list.
 func (sms StagedMetadatas) IsDefault() bool {
 	return len(sms) == 1 && sms[0].IsDefault()
@@ -396,6 +421,49 @@ func (sms StagedMetadatas) IsDropPolicyApplied() bool {
 	return len(sms) == 1 && sms[0].IsDropPolicyApplied()
 }
 
+// ApplyOrRemoveDropPolicies applies or removes any drop policies staged
+// metadatas, if effective then just a single drop pipeline staged metadata
+// is returned otherwise if not effective it removes in each staged metadata
+// the drop policy from all pipelines and retains only non-drop policy
+// effective staged metadatas.
+func (sms StagedMetadatas) ApplyOrRemoveDropPolicies() (
+	StagedMetadatas,
+	ApplyOrRemoveDropPoliciesResult,
+) {
+	if len(sms) == 0 {
+		return sms, RemovedIneffectiveDropPoliciesResult
+	}
+
+	nonDropStagedMetadatas := 0
+	result := sms
+	for i := len(result) - 1; i >= 0; i-- {
+		var applyOrRemoveResult ApplyOrRemoveDropPoliciesResult
+		sms[i].Pipelines, applyOrRemoveResult = sms[i].Pipelines.ApplyOrRemoveDropPolicies()
+
+		switch applyOrRemoveResult {
+		case AppliedEffectiveDropPolicyResult:
+			// Remove by moving to tail and decrementing length so we can do in
+			// place to avoid allocations of a new slice
+			if lastElem := i == len(result)-1; lastElem {
+				result = result[0:i]
+			} else {
+				result = append(result[0:i], result[i+1:]...)
+			}
+		default:
+			// Not an effective drop staged metadata
+			nonDropStagedMetadatas++
+		}
+	}
+
+	if nonDropStagedMetadatas == 0 {
+		// If there were no non-drop staged metadatas, then just return the
+		// canonical drop staged metadatas
+		return DropStagedMetadatas, AppliedEffectiveDropPolicyResult
+	}
+
+	return result, RemovedIneffectiveDropPoliciesResult
+}
+
 // ToProto converts the staged metadatas to a protobuf message in place.
 func (sms StagedMetadatas) ToProto(pb *metricpb.StagedMetadatas) error {
 	numMetadatas := len(sms)

metadata/metadata_test.go

@@ -1164,8 +1164,7 @@ func TestApplyOrRemoveDropPoliciesDropMust(t *testing.T) {
 	}
 	output, result := input.ApplyOrRemoveDropPolicies()
 	require.Equal(t, AppliedEffectiveDropPolicyResult, result)
-	require.Equal(t, 1, len(output))
-	require.True(t, output[0].Equal(DropPipelineMetadata))
+	require.True(t, output.Equal(DropPipelineMetadatas))
 }
 
 func TestApplyOrRemoveDropPoliciesDropIfOnlyMatchEffective(t *testing.T) {
@@ -1178,8 +1177,7 @@ func TestApplyOrRemoveDropPoliciesDropIfOnlyMatchEffective(t *testing.T) {
 	}
 	output, result := input.ApplyOrRemoveDropPolicies()
 	require.Equal(t, AppliedEffectiveDropPolicyResult, result)
-	require.Equal(t, 1, len(output))
-	require.True(t, output[0].Equal(DropPipelineMetadata))
+	require.True(t, output.Equal(DropPipelineMetadatas))
 }
 
 func TestApplyOrRemoveDropPoliciesDropIfOnlyMatchMiddleIneffective(t *testing.T) {
@@ -1226,9 +1224,7 @@ func TestApplyOrRemoveDropPoliciesDropIfOnlyMatchMiddleIneffective(t *testing.T)
 
 				output, result := input.ApplyOrRemoveDropPolicies()
 				require.Equal(t, RemovedIneffectiveDropPoliciesResult, result)
-				require.Equal(t, 2, len(output))
-				require.True(t, output[0].Equal(validRules[0]))
-				require.True(t, output[1].Equal(validRules[1]))
+				require.True(t, output.Equal(validRules))
 			})
 	}
 }
@@ -1254,7 +1250,87 @@ func TestApplyOrRemoveDropPoliciesDropIfOnlyMatchNone(t *testing.T) {
 	}
 	output, result := input.ApplyOrRemoveDropPolicies()
 	require.Equal(t, RemovedIneffectiveDropPoliciesResult, result)
-	require.Equal(t, 2, len(output))
-	require.True(t, output[0].Equal(input[0]))
-	require.True(t, output[1].Equal(input[1]))
+	require.True(t, output.Equal(input))
+}
+
+func TestStagedMetadatasApplyOrRemoveDropPoliciesRemovingAnyDropStagedMetadata(t *testing.T) {
+	validStagedMetadatas := StagedMetadatas{
+		StagedMetadata{
+			Metadata: Metadata{Pipelines: PipelineMetadatas{
+				{
+					AggregationID: aggregation.MustCompressTypes(aggregation.Sum),
+					StoragePolicies: []policy.StoragePolicy{
+						policy.NewStoragePolicy(time.Second, xtime.Second, time.Hour),
+						policy.NewStoragePolicy(time.Minute, xtime.Minute, 12*time.Hour),
+					},
+					DropPolicy: policy.DropNone,
+				},
+			}},
+		},
+		StagedMetadata{
+			Metadata: Metadata{Pipelines: PipelineMetadatas{
+				{
+					AggregationID: aggregation.MustCompressTypes(aggregation.Sum),
+					StoragePolicies: []policy.StoragePolicy{
+						policy.NewStoragePolicy(time.Minute, xtime.Minute, 12*time.Hour),
+						policy.NewStoragePolicy(10*time.Minute, xtime.Minute, 24*time.Hour),
+					},
+					DropPolicy: policy.DropNone,
+				},
+			}},
+		},
+	}
+
+	// Run test for every single insertion point
+	for i := 0; i < len(validStagedMetadatas)+1; i++ {
+		t.Run(fmt.Sprintf("test insert drop if only rule at %d", i),
+			func(t *testing.T) {
+				var (
+					copy  = append(StagedMetadatas(nil), validStagedMetadatas...)
+					input StagedMetadatas
+				)
+				for j := 0; j < len(validStagedMetadatas)+1; j++ {
+					if j == i {
+						// Insert the drop if only match rule at this position
+						input = append(input, DropStagedMetadata)
+					} else {
+						input = append(input, copy[0])
+						copy = copy[1:]
+					}
+				}
+
+				output, result := input.ApplyOrRemoveDropPolicies()
+				require.Equal(t, RemovedIneffectiveDropPoliciesResult, result)
+				require.True(t, output.Equal(validStagedMetadatas))
+			})
+	}
+}
+
+func TestStagedMetadatasApplyOrRemoveDropPoliciesApplyingDropStagedMetadata(t *testing.T) {
+	// Check compacts together
+	metadatas, result := StagedMetadatas{
+		DropStagedMetadata,
+		DropStagedMetadata,
+	}.ApplyOrRemoveDropPolicies()
+
+	require.True(t, metadatas.Equal(DropStagedMetadatas))
+	require.Equal(t, AppliedEffectiveDropPolicyResult, result)
+
+	// Check single also returns as expected
+	metadatas, result = StagedMetadatas{
+		DropStagedMetadata,
+	}.ApplyOrRemoveDropPolicies()
+
+	require.True(t, metadatas.Equal(DropStagedMetadatas))
+	require.Equal(t, AppliedEffectiveDropPolicyResult, result)
+}
+
+func TestStagedMetadatasApplyOrRemoveDropPoliciesWithNoStagedMetadatasIsNoOp(t *testing.T) {
+	metadatas, result := StagedMetadatas{}.ApplyOrRemoveDropPolicies()
+	require.Equal(t, 0, len(metadatas))
+	require.Equal(t, RemovedIneffectiveDropPoliciesResult, result)
+}
+
+func TestDropStagedMetadatasReturnsIsDropPolicyAppliedTrue(t *testing.T) {
+	require.True(t, DropStagedMetadatas.IsDropPolicyApplied())
 }

policy/drop_policy.go

@@ -47,7 +47,7 @@ var validDropPolicies = []DropPolicy{
 
 // IsDefault returns whether the drop policy is the default drop none policy.
 func (p DropPolicy) IsDefault() bool {
-	return p == DropNone
+	return p == DefaultDropPolicy
 }
 
 func (p DropPolicy) String() string {
@@ -57,7 +57,7 @@ func (p DropPolicy) String() string {
 	case DropMust:
 		return "drop_must"
 	case DropIfOnlyMatch:
-		return "drop_except_if_other_match"
+		return "drop_if_only_match"
 	}
 	return DropNone.String()
 }

policy/drop_policy_test.go

@@ -0,0 +1,56 @@
+// Copyright (c) 2018 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 policy
+
+import (
+	"fmt"
+	"math"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestDropPolicyIsDefault(t *testing.T) {
+	assert.True(t, DefaultDropPolicy.IsDefault())
+}
+
+func TestDropPolicyString(t *testing.T) {
+	for _, policy := range validDropPolicies {
+		switch policy {
+		case DropNone:
+			assert.Equal(t, "drop_none", policy.String())
+		case DropMust:
+			assert.Equal(t, "drop_must", policy.String())
+		case DropIfOnlyMatch:
+			assert.Equal(t, "drop_if_only_match", policy.String())
+		default:
+			assert.Fail(t, fmt.Sprintf("unknown policy: value=%d, string=%s",
+				int(policy), policy.String()))
+		}
+	}
+}
+
+func TestDropPolicyIsValid(t *testing.T) {
+	for _, policy := range validDropPolicies {
+		assert.True(t, policy.IsValid())
+	}
+	assert.False(t, DropPolicy(math.MaxUint64).IsValid())
+}