Merge branch 'master' into r/drop-mapping-rule

metadata/metadata.go

@@ -33,7 +33,7 @@ var (
 	DefaultPipelineMetadata PipelineMetadata
 
 	// DefaultPipelineMetadatas is a default list of pipeline metadatas.
-	DefaultPipelineMetadatas = []PipelineMetadata{DefaultPipelineMetadata}
+	DefaultPipelineMetadatas = PipelineMetadatas{DefaultPipelineMetadata}
 
 	// DefaultMetadata is a default metadata.
 	DefaultMetadata = Metadata{Pipelines: DefaultPipelineMetadatas}
@@ -91,6 +91,15 @@ func (m PipelineMetadata) IsDropPolicyApplied() bool {
 		!m.DropPolicy.IsDefault()
 }
 
+// Clone clones the pipeline metadata.
+func (m PipelineMetadata) Clone() PipelineMetadata {
+	return PipelineMetadata{
+		AggregationID:   m.AggregationID,
+		StoragePolicies: m.StoragePolicies.Clone(),
+		Pipeline:        m.Pipeline.Clone(),
+	}
+}
+
 // ToProto converts the pipeline metadata to a protobuf message in place.
 func (m PipelineMetadata) ToProto(pb *metricpb.PipelineMetadata) error {
 	if err := m.AggregationID.ToProto(&pb.AggregationId); err != nil {
@@ -137,9 +146,34 @@ func (m *PipelineMetadata) FromProto(pb metricpb.PipelineMetadata) error {
 	return nil
 }
 
+// PipelineMetadatas is a list of pipeline metadatas.
+type PipelineMetadatas []PipelineMetadata
+
+// Equal returns true if two pipline metadatas are considered equal.
+func (metadatas PipelineMetadatas) Equal(other PipelineMetadatas) bool {
+	if len(metadatas) != len(other) {
+		return false
+	}
+	for i := 0; i < len(metadatas); i++ {
+		if !metadatas[i].Equal(other[i]) {
+			return false
+		}
+	}
+	return true
+}
+
+// Clone clones the list of pipeline metadatas.
+func (metadatas PipelineMetadatas) Clone() PipelineMetadatas {
+	cloned := make(PipelineMetadatas, 0, len(metadatas))
+	for i := 0; i < len(metadatas); i++ {
+		cloned = append(cloned, metadatas[i].Clone())
+	}
+	return cloned
+}
+
 // Metadata represents the metadata associated with a metric.
 type Metadata struct {
-	Pipelines []PipelineMetadata `json:"pipelines"`
+	Pipelines PipelineMetadatas `json:"pipelines"`
 }
 
 // IsDefault returns whether this is the default metadata.
@@ -175,7 +209,7 @@ func (m *Metadata) FromProto(pb metricpb.Metadata) error {
 	if cap(m.Pipelines) >= numPipelines {
 		m.Pipelines = m.Pipelines[:numPipelines]
 	} else {
-		m.Pipelines = make([]PipelineMetadata, numPipelines)
+		m.Pipelines = make(PipelineMetadatas, numPipelines)
 	}
 	for i := 0; i < numPipelines; i++ {
 		if err := m.Pipelines[i].FromProto(pb.Pipelines[i]); err != nil {

metadata/metadata_test.go

@@ -810,6 +810,18 @@ func TestForwardMetadataFromProtoBadMetadataProto(t *testing.T) {
 	require.Error(t, res.FromProto(testBadForwardMetadataProto))
 }
 
+func TestPipelineMetadataClone(t *testing.T) {
+	cloned1 := testLargePipelineMetadata.Clone()
+	cloned2 := testLargePipelineMetadata.Clone()
+	require.True(t, cloned1.Equal(testLargePipelineMetadata))
+	require.True(t, cloned2.Equal(testLargePipelineMetadata))
+
+	// Assert that modifying the clone does not mutate the original pipeline metadata.
+	cloned1.StoragePolicies[0] = policy.MustParseStoragePolicy("1h:1h")
+	require.False(t, cloned1.Equal(testLargePipelineMetadata))
+	require.True(t, cloned2.Equal(testLargePipelineMetadata))
+}
+
 func TestPipelineMetadataToProto(t *testing.T) {
 	inputs := []struct {
 		sequence []PipelineMetadata
@@ -917,6 +929,22 @@ func TestPipelineMetadataFromProtoBadMetadataProto(t *testing.T) {
 	require.Error(t, res.FromProto(testBadPipelineMetadataProto))
 }
 
+func TestPipelineMetadatasClone(t *testing.T) {
+	input := PipelineMetadatas{
+		testSmallPipelineMetadata,
+		testLargePipelineMetadata,
+	}
+	cloned1 := input.Clone()
+	cloned2 := input.Clone()
+	require.True(t, cloned1.Equal(input))
+	require.True(t, cloned2.Equal(input))
+
+	// Assert that modifying the clone does not mutate the original pipeline metadata.
+	cloned1[0].StoragePolicies[0] = policy.MustParseStoragePolicy("1h:1h")
+	require.False(t, cloned1.Equal(input))
+	require.True(t, cloned2.Equal(input))
+}
+
 func TestStagedMetadatasToProto(t *testing.T) {
 	inputs := []struct {
 		sequence []StagedMetadatas

rules/active_ruleset.go

@@ -235,6 +235,12 @@ func (as *activeRuleSet) mappingsForNonRollupID(
 		}
 		pipelines = append(pipelines, pipeline)
 	}
+	// NB: The pipeline list should never be empty as the resulting pipelines are
+	// used to determine how the *existing* ID is aggregated and retained. If there
+	// are no rule match, the default pipeline list is used.
+	if len(pipelines) == 0 {
+		pipelines = metadata.DefaultPipelineMetadatas.Clone()
+	}
 	return mappingResults{
 		forExistingID: ruleMatchResults{cutoverNanos: cutoverNanos, pipelines: pipelines},
 	}
@@ -491,7 +497,7 @@ func (as *activeRuleSet) reverseMappingsForNonRollupID(
 	at aggregation.Type,
 ) (metadata.StagedMetadata, bool) {
 	mappingRes := as.mappingsForNonRollupID(id, timeNanos).forExistingID
-	filteredPipelines := filteredPipelinesWithAggregationType(mappingRes.resolvedPipelines(), mt, at, as.aggTypeOpts)
+	filteredPipelines := filteredPipelinesWithAggregationType(mappingRes.pipelines, mt, at, as.aggTypeOpts)
 	if len(filteredPipelines) == 0 {
 		return metadata.DefaultStagedMetadata, false
 	}

rules/active_ruleset_test.go

@@ -38,11 +38,13 @@ import (
 
 	"github.com/google/go-cmp/cmp"
 	"github.com/google/go-cmp/cmp/cmpopts"
-	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
 
 var (
+	testStagedMetadatasCmptOpts = []cmp.Option{
+		cmpopts.EquateEmpty(),
+	}
 	testIDWithMetadatasCmpOpts = []cmp.Option{
 		cmpopts.EquateEmpty(),
 	}
@@ -521,11 +523,12 @@ func TestActiveRuleSetForwardMatchWithMappingRules(t *testing.T) {
 		nil,
 		aggregation.NewTypesOptions(),
 	)
+
 	for i, input := range inputs {
 		t.Run(fmt.Sprintf("input %d", i), func(t *testing.T) {
 			res := as.ForwardMatch(b(input.id), input.matchFrom, input.matchTo)
 			require.Equal(t, input.expireAtNanos, res.expireAtNanos)
-			require.Equal(t, input.forExistingIDResult, res.ForExistingIDAt(0))
+			require.True(t, cmp.Equal(input.forExistingIDResult, res.ForExistingIDAt(0), testStagedMetadatasCmptOpts...))
 			require.Equal(t, 0, res.NumNewRollupIDs())
 		})
 	}
@@ -544,6 +547,7 @@ func TestActiveRuleSetForwardMatchWithRollupRules(t *testing.T) {
 					Tombstoned:   false,
 					Metadata: metadata.Metadata{
 						Pipelines: []metadata.PipelineMetadata{
+							metadata.DefaultPipelineMetadata,
 							{
 								AggregationID: aggregation.MustCompressTypes(aggregation.Sum),
 								StoragePolicies: policy.StoragePolicies{
@@ -711,6 +715,7 @@ func TestActiveRuleSetForwardMatchWithRollupRules(t *testing.T) {
 					Tombstoned:   false,
 					Metadata: metadata.Metadata{
 						Pipelines: []metadata.PipelineMetadata{
+							metadata.DefaultPipelineMetadata,
 							{
 								AggregationID: aggregation.MustCompressTypes(aggregation.Sum),
 								StoragePolicies: policy.StoragePolicies{
@@ -736,6 +741,7 @@ func TestActiveRuleSetForwardMatchWithRollupRules(t *testing.T) {
 					Tombstoned:   false,
 					Metadata: metadata.Metadata{
 						Pipelines: []metadata.PipelineMetadata{
+							metadata.DefaultPipelineMetadata,
 							{
 								AggregationID: aggregation.MustCompressTypes(aggregation.Sum),
 								StoragePolicies: policy.StoragePolicies{
@@ -783,6 +789,7 @@ func TestActiveRuleSetForwardMatchWithRollupRules(t *testing.T) {
 					Tombstoned:   false,
 					Metadata: metadata.Metadata{
 						Pipelines: []metadata.PipelineMetadata{
+							metadata.DefaultPipelineMetadata,
 							{
 								AggregationID: aggregation.MustCompressTypes(aggregation.Sum),
 								StoragePolicies: policy.StoragePolicies{
@@ -830,6 +837,7 @@ func TestActiveRuleSetForwardMatchWithRollupRules(t *testing.T) {
 					Tombstoned:   false,
 					Metadata: metadata.Metadata{
 						Pipelines: []metadata.PipelineMetadata{
+							metadata.DefaultPipelineMetadata,
 							{
 								AggregationID: aggregation.DefaultID,
 								StoragePolicies: policy.StoragePolicies{
@@ -881,6 +889,7 @@ func TestActiveRuleSetForwardMatchWithRollupRules(t *testing.T) {
 					Tombstoned:   false,
 					Metadata: metadata.Metadata{
 						Pipelines: []metadata.PipelineMetadata{
+							metadata.DefaultPipelineMetadata,
 							{
 								AggregationID: aggregation.DefaultID,
 								StoragePolicies: policy.StoragePolicies{
@@ -932,6 +941,7 @@ func TestActiveRuleSetForwardMatchWithRollupRules(t *testing.T) {
 					Tombstoned:   false,
 					Metadata: metadata.Metadata{
 						Pipelines: []metadata.PipelineMetadata{
+							metadata.DefaultPipelineMetadata,
 							{
 								AggregationID: aggregation.DefaultID,
 								StoragePolicies: policy.StoragePolicies{
@@ -961,6 +971,7 @@ func TestActiveRuleSetForwardMatchWithRollupRules(t *testing.T) {
 					Tombstoned:   false,
 					Metadata: metadata.Metadata{
 						Pipelines: []metadata.PipelineMetadata{
+							metadata.DefaultPipelineMetadata,
 							{
 								AggregationID: aggregation.DefaultID,
 								StoragePolicies: policy.StoragePolicies{
@@ -1284,11 +1295,12 @@ func TestActiveRuleSetForwardMatchWithRollupRules(t *testing.T) {
 		nil,
 		aggregation.NewTypesOptions(),
 	)
+
 	for i, input := range inputs {
 		t.Run(fmt.Sprintf("input %d", i), func(t *testing.T) {
 			res := as.ForwardMatch(b(input.id), input.matchFrom, input.matchTo)
 			require.Equal(t, input.expireAtNanos, res.expireAtNanos)
-			require.Equal(t, input.forExistingIDResult, res.ForExistingIDAt(0))
+			require.True(t, cmp.Equal(input.forExistingIDResult, res.ForExistingIDAt(0), testStagedMetadatasCmptOpts...))
 			require.Equal(t, len(input.forNewRollupIDsResult), res.NumNewRollupIDs())
 			for i := 0; i < len(input.forNewRollupIDsResult); i++ {
 				rollup := res.ForNewRollupIDsAt(i, 0)
@@ -2381,15 +2393,16 @@ func TestActiveRuleSetForwardMatchWithMappingRulesAndRollupRules(t *testing.T) {
 		nil,
 		aggregation.NewTypesOptions(),
 	)
+
 	for i, input := range inputs {
 		t.Run(fmt.Sprintf("input %d", i), func(t *testing.T) {
 			res := as.ForwardMatch(b(input.id), input.matchFrom, input.matchTo)
-			assert.Equal(t, input.expireAtNanos, res.expireAtNanos)
-			assert.Equal(t, input.forExistingIDResult, res.ForExistingIDAt(0))
-			require.Equal(t, len(input.forNewRollupIDsResult), res.NumNewRollupIDs(), "failed num rollup ids expected=%d, actual=%d", len(input.forNewRollupIDsResult), res.NumNewRollupIDs())
+			require.Equal(t, input.expireAtNanos, res.expireAtNanos)
+			require.True(t, cmp.Equal(input.forExistingIDResult, res.ForExistingIDAt(0), testStagedMetadatasCmptOpts...))
+			require.Equal(t, len(input.forNewRollupIDsResult), res.NumNewRollupIDs())
 			for i := 0; i < len(input.forNewRollupIDsResult); i++ {
 				rollup := res.ForNewRollupIDsAt(i, 0)
-				assert.True(t, cmp.Equal(input.forNewRollupIDsResult[i], rollup, testIDWithMetadatasCmpOpts...))
+				require.True(t, cmp.Equal(input.forNewRollupIDsResult[i], rollup, testIDWithMetadatasCmpOpts...))
 			}
 		})
 	}
@@ -2807,7 +2820,7 @@ func TestActiveRuleSetReverseMatchWithMappingRulesForNonRollupID(t *testing.T) {
 	for _, input := range inputs {
 		res := as.ReverseMatch(b(input.id), input.matchFrom, input.matchTo, input.metricType, input.aggregationType)
 		require.Equal(t, input.expireAtNanos, res.expireAtNanos)
-		require.Equal(t, input.forExistingIDResult, res.ForExistingIDAt(0))
+		require.True(t, cmp.Equal(input.forExistingIDResult, res.ForExistingIDAt(0), testStagedMetadatasCmptOpts...))
 	}
 }
 

rules/validator/config.go

@@ -59,28 +59,46 @@ func (c Configuration) NewValidator(
 	if err != nil {
 		return nil, err
 	}
+	opts := c.newValidatorOptions(nsValidator)
+	return NewValidator(opts), nil
+}
+
+func (c Configuration) newValidatorOptions(
+	nsValidator namespace.Validator,
+) Options {
 	opts := NewOptions().
 		SetNamespaceValidator(nsValidator).
 		SetRequiredRollupTags(c.RequiredRollupTags).
 		SetMetricTypesFn(c.MetricTypes.NewMetricTypesFn()).
-		SetDefaultAllowedStoragePolicies(c.Policies.DefaultAllowed.StoragePolicies).
-		SetDefaultAllowedFirstLevelAggregationTypes(c.Policies.DefaultAllowed.FirstLevelAggregationTypes).
-		SetDefaultAllowedNonFirstLevelAggregationTypes(c.Policies.DefaultAllowed.NonFirstLevelAggregationTypes).
 		SetTagNameInvalidChars(toRunes(c.TagNameInvalidChars)).
 		SetMetricNameInvalidChars(toRunes(c.MetricNameInvalidChars))
+	if c.Policies.DefaultAllowed.StoragePolicies != nil {
+		opts = opts.SetDefaultAllowedStoragePolicies(*c.Policies.DefaultAllowed.StoragePolicies)
+	}
+	if c.Policies.DefaultAllowed.FirstLevelAggregationTypes != nil {
+		opts = opts.SetDefaultAllowedFirstLevelAggregationTypes(*c.Policies.DefaultAllowed.FirstLevelAggregationTypes)
+	}
+	if c.Policies.DefaultAllowed.NonFirstLevelAggregationTypes != nil {
+		opts = opts.SetDefaultAllowedNonFirstLevelAggregationTypes(*c.Policies.DefaultAllowed.NonFirstLevelAggregationTypes)
+	}
 	for _, override := range c.Policies.Overrides {
-		opts = opts.
-			SetAllowedStoragePoliciesFor(override.Type, override.Allowed.StoragePolicies).
-			SetAllowedFirstLevelAggregationTypesFor(override.Type, override.Allowed.FirstLevelAggregationTypes).
-			SetAllowedNonFirstLevelAggregationTypesFor(override.Type, override.Allowed.NonFirstLevelAggregationTypes)
+		if override.Allowed.StoragePolicies != nil {
+			opts = opts.SetAllowedStoragePoliciesFor(override.Type, *override.Allowed.StoragePolicies)
+		}
+		if override.Allowed.FirstLevelAggregationTypes != nil {
+			opts = opts.SetAllowedFirstLevelAggregationTypesFor(override.Type, *override.Allowed.FirstLevelAggregationTypes)
+		}
+		if override.Allowed.NonFirstLevelAggregationTypes != nil {
+			opts = opts.SetAllowedNonFirstLevelAggregationTypesFor(override.Type, *override.Allowed.NonFirstLevelAggregationTypes)
+		}
 	}
 	if c.MaxTransformationDerivativeOrder != nil {
 		opts = opts.SetMaxTransformationDerivativeOrder(*c.MaxTransformationDerivativeOrder)
 	}
 	if c.MaxRollupLevels != nil {
 		opts = opts.SetMaxRollupLevels(*c.MaxRollupLevels)
 	}
-	return NewValidator(opts), nil
+	return opts
 }
 
 type namespaceValidatorConfiguration struct {
@@ -152,9 +170,9 @@ type policiesOverrideConfiguration struct {
 
 // policiesConfiguration is the configuration for storage policies and aggregation types.
 type policiesConfiguration struct {
-	StoragePolicies               []policy.StoragePolicy `yaml:"storagePolicies"`
-	FirstLevelAggregationTypes    []aggregation.Type     `yaml:"firstLevelAggregationTypes"`
-	NonFirstLevelAggregationTypes []aggregation.Type     `yaml:"nonFirstLevelAggregationTypes"`
+	StoragePolicies               *[]policy.StoragePolicy `yaml:"storagePolicies"`
+	FirstLevelAggregationTypes    *[]aggregation.Type     `yaml:"firstLevelAggregationTypes"`
+	NonFirstLevelAggregationTypes *[]aggregation.Type     `yaml:"nonFirstLevelAggregationTypes"`
 }
 
 func toRunes(s string) []rune {