Provide options to enforce required rollup tags

rules/validator.go

@@ -100,8 +100,9 @@ func (v *validator) validateMappingRules(mrv map[string]*MappingRuleView) error
 
 func (v *validator) validateRollupRules(rrv map[string]*RollupRuleView) error {
 	var (
-		namesSeen   = make(map[string]struct{}, len(rrv))
-		targetsSeen = make([]RollupTarget, 0, len(rrv))
+		namesSeen    = make(map[string]struct{}, len(rrv))
+		targetsSeen  = make([]RollupTarget, 0, len(rrv))
+		requiredTags = v.opts.RequiredRollupTags()
 	)
 	for _, view := range rrv {
 		// Validate that no rules with the same name exist.
@@ -124,19 +125,30 @@ func (v *validator) validateRollupRules(rrv map[string]*RollupRuleView) error {
 			return fmt.Errorf("rollup rule %s does not match any allowed metric types, filter=%v", view.Name, view.Filters)
 		}
 
-		// Validate that the policies are valid.
-		for _, t := range types {
-			for _, target := range view.Targets {
+		for _, target := range view.Targets {
+			// Validate that the rollup tags contain the list of required rollup tags.
+			if len(requiredTags) > 0 {
+				rollupTags := make(map[string]struct{}, len(target.Tags))
+				for _, tag := range target.Tags {
+					rollupTags[tag] = struct{}{}
+				}
+				for _, requiredTag := range requiredTags {
+					if _, exists := rollupTags[requiredTag]; !exists {
+						return fmt.Errorf("rollup rule %s does not have required rollup tag: %s, provided rollup tags are %v", view.Name, requiredTag, target.Tags)
+					}
+				}
+			}
+
+			// Validate that the policies are valid.
+			for _, t := range types {
 				for _, p := range target.Policies {
 					if err := v.validatePolicy(t, p); err != nil {
 						return err
 					}
 				}
 			}
-		}
 
-		// Validate that there are no conflicting rollup targets.
-		for _, target := range view.Targets {
+			// Validate that there are no conflicting rollup targets.
 			current := target.rollupTarget()
 			for _, seenTarget := range targetsSeen {
 				if current.sameTransform(seenTarget) {

rules/validator_options.go

@@ -50,6 +50,12 @@ type ValidatorOptions interface {
 	// MetricTypesFn returns the metric types function.
 	MetricTypesFn() MetricTypesFn
 
+	// SetRequiredRollupTags sets the list of required rollup tags.
+	SetRequiredRollupTags(value []string) ValidatorOptions
+
+	// RequiredRollupTags returns the list of required rollup tags.
+	RequiredRollupTags() []string
+
 	// IsAllowedStoragePolicyFor determines whether a given storage policy is allowed for the
 	// given metric type.
 	IsAllowedStoragePolicyFor(t metric.Type, p policy.StoragePolicy) bool
@@ -68,6 +74,7 @@ type validatorOptions struct {
 	defaultAllowedStoragePolicies        map[policy.StoragePolicy]struct{}
 	defaultAllowedCustomAggregationTypes map[policy.AggregationType]struct{}
 	metricTypesFn                        MetricTypesFn
+	requiredRollupTags                   []string
 	metadatasByType                      map[metric.Type]validationMetadata
 }
 
@@ -111,6 +118,18 @@ func (o *validatorOptions) MetricTypesFn() MetricTypesFn {
 	return o.metricTypesFn
 }
 
+func (o *validatorOptions) SetRequiredRollupTags(value []string) ValidatorOptions {
+	requiredRollupTags := make([]string, len(value))
+	copy(requiredRollupTags, value)
+	opts := *o
+	opts.requiredRollupTags = requiredRollupTags
+	return &opts
+}
+
+func (o *validatorOptions) RequiredRollupTags() []string {
+	return o.requiredRollupTags
+}
+
 func (o *validatorOptions) IsAllowedStoragePolicyFor(t metric.Type, p policy.StoragePolicy) bool {
 	if metadata, exists := o.metadatasByType[t]; exists {
 		_, found := metadata.allowedStoragePolicies[p]

rules/validator_test.go

@@ -184,6 +184,13 @@ func TestValidatorValidateRollupRuleInvalidMetricType(t *testing.T) {
 	require.Error(t, ruleSet.Validate(validator))
 }
 
+func TestValidatorValidateRollupRuleMissingRequiredTag(t *testing.T) {
+	requiredRollupTags := []string{"requiredTag"}
+	ruleSet := testRuleSetWithRollupRules(t, testMissingRequiredTagRollupRulesConfig())
+	validator := NewValidator(testValidatorOptions().SetRequiredRollupTags(requiredRollupTags))
+	require.Error(t, ruleSet.Validate(validator))
+}
+
 func TestValidatorValidateRollupRulePolicy(t *testing.T) {
 	testStoragePolicies := []policy.StoragePolicy{
 		policy.MustParseStoragePolicy("10s:1d"),
@@ -455,6 +462,26 @@ func testInvalidMetricTypeRollupRulesConfig() []*schema.RollupRule {
 	}
 }
 
+func testMissingRequiredTagRollupRulesConfig() []*schema.RollupRule {
+	return []*schema.RollupRule{
+		&schema.RollupRule{
+			Uuid: "rollupRule1",
+			Snapshots: []*schema.RollupRuleSnapshot{
+				&schema.RollupRuleSnapshot{
+					Name:       "snapshot1",
+					Tombstoned: false,
+					Targets: []*schema.RollupTarget{
+						&schema.RollupTarget{
+							Name: "rName1",
+							Tags: []string{"rtagName1", "rtagName2"},
+						},
+					},
+				},
+			},
+		},
+	}
+}
+
 func testPolicyResolutionRollupRulesConfig() []*schema.RollupRule {
 	return []*schema.RollupRule{
 		&schema.RollupRule{