address comments, naming sweep

metric/aggregated/types.go

@@ -74,25 +74,25 @@ type RawMetric interface {
 	Reset(data []byte)
 }
 
-// MetricWithPolicy is a metric with applicable policy.
-type MetricWithPolicy struct {
+// MetricWithStoragePolicy is a metric with applicable storage policy.
+type MetricWithStoragePolicy struct {
 	Metric
 	policy.StoragePolicy
 }
 
-// String is the string representation of a metric with policy.
-func (mp MetricWithPolicy) String() string {
+// String is the string representation of a metric with storage policy.
+func (mp MetricWithStoragePolicy) String() string {
 	return fmt.Sprintf("{metric:%s,policy:%s}", mp.Metric.String(), mp.StoragePolicy.String())
 }
 
-// ChunkedMetricWithPolicy is a chunked metric with applicable policy.
-type ChunkedMetricWithPolicy struct {
+// ChunkedMetricWithStoragePolicy is a chunked metric with applicable storage policy.
+type ChunkedMetricWithStoragePolicy struct {
 	ChunkedMetric
 	policy.StoragePolicy
 }
 
-// RawMetricWithPolicy is a raw metric with applicable policy.
-type RawMetricWithPolicy struct {
+// RawMetricWithStoragePolicy is a raw metric with applicable storage policy.
+type RawMetricWithStoragePolicy struct {
 	RawMetric
 	policy.StoragePolicy
 }

policy/aggregation_id_compress.go

@@ -59,8 +59,11 @@ func (c *aggregationIDCompressor) Compress(aggTypes AggregationTypes) (Aggregati
 		}
 		c.bs.Set(uint(aggType))
 	}
+
 	codes := c.bs.Bytes()
 	var id AggregationID
+	// NB(cw) it's guaranteed that len(id) == len(codes) == AggregationIDLen, we need to copy
+	// the words in bitset out because the bitset contains a slice internally
 	for i := 0; i < AggregationIDLen; i++ {
 		id[i] = codes[i]
 	}
@@ -84,6 +87,8 @@ func NewAggregationTypeDecompressor() AggregationIDDecompressor {
 }
 
 func (c *aggregationIDDecompressor) Decompress(id AggregationID) (AggregationTypes, error) {
+	// NB(cw) it's guaranteed that len(c.buf) == len(id) == AggregationIDLen, we need to copy
+	// the words from id into a slice to be used in bitset
 	for i := range id {
 		c.buf[i] = id[i]
 	}

policy/aggregation_type.go

@@ -44,6 +44,7 @@ const (
 	P99
 	P999
 	P9999
+
 	totalAggregationTypes = iota
 )
 
@@ -56,36 +57,37 @@ const (
 )
 
 var (
+	emptyStruct struct{}
 	// DefaultAggregationTypes is a default list of aggregation types.
 	DefaultAggregationTypes AggregationTypes
 
 	// DefaultAggregationID is a default AggregationID.
 	DefaultAggregationID AggregationID
 
 	// ValidAggregationTypes is the list of all the valid aggregation types
-	ValidAggregationTypes = []AggregationType{
-		Last,
-		Lower,
-		Upper,
-		Mean,
-		Median,
-		Count,
-		Sum,
-		SumSq,
-		Stdev,
-		P50,
-		P95,
-		P99,
-		P999,
-		P9999,
+	ValidAggregationTypes = map[AggregationType]struct{}{
+		Last:   emptyStruct,
+		Lower:  emptyStruct,
+		Upper:  emptyStruct,
+		Mean:   emptyStruct,
+		Median: emptyStruct,
+		Count:  emptyStruct,
+		Sum:    emptyStruct,
+		SumSq:  emptyStruct,
+		Stdev:  emptyStruct,
+		P50:    emptyStruct,
+		P95:    emptyStruct,
+		P99:    emptyStruct,
+		P999:   emptyStruct,
+		P9999:  emptyStruct,
 	}
 
 	aggregationTypeStringMap map[string]AggregationType
 )
 
 func init() {
 	aggregationTypeStringMap = make(map[string]AggregationType, totalAggregationTypes)
-	for _, aggType := range ValidAggregationTypes {
+	for aggType := range ValidAggregationTypes {
 		aggregationTypeStringMap[aggType.String()] = aggType
 	}
 }
@@ -104,7 +106,8 @@ func NewAggregationTypeFromSchema(input schema.AggregationType) (AggregationType
 
 // IsValid checks if an AggregationType is valid.
 func (a AggregationType) IsValid() bool {
-	return a > 0 && a < totalAggregationTypes
+	_, ok := ValidAggregationTypes[a]
+	return ok
 }
 
 // IsValidForGauge if an AggregationType is valid for Gauge.
@@ -138,9 +141,12 @@ func (a AggregationType) IsValidForTimer() bool {
 }
 
 // ParseAggregationType parses an aggregation type.
-func ParseAggregationType(str string) (AggregationType, bool) {
+func ParseAggregationType(str string) (AggregationType, error) {
 	aggType, ok := aggregationTypeStringMap[str]
-	return aggType, ok
+	if !ok {
+		return Unknown, fmt.Errorf("invalid aggregation type: %s", str)
+	}
+	return aggType, nil
 }
 
 // AggregationTypes is a list of AggregationTypes.
@@ -212,9 +218,9 @@ func ParseAggregationTypes(str string) (AggregationTypes, error) {
 	parts := strings.Split(str, aggregationTypesSeparator)
 	res := make(AggregationTypes, len(parts))
 	for i := range parts {
-		aggType, ok := ParseAggregationType(parts[i])
-		if !ok {
-			return nil, fmt.Errorf("invalid aggregation type: %s", parts[i])
+		aggType, err := ParseAggregationType(parts[i])
+		if err != nil {
+			return nil, err
 		}
 		res[i] = aggType
 	}

policy/policy.go

@@ -39,15 +39,15 @@ var (
 	errInvalidPolicyString = errors.New("invalid policy string")
 )
 
-// Policy contains a policy and a list of custom aggregation types.
+// Policy contains a storage policy and a list of custom aggregation types.
 type Policy struct {
-	sp    StoragePolicy
-	aggID AggregationID
+	StoragePolicy
+	AggregationID
 }
 
 // NewPolicy creates a policy.
 func NewPolicy(sp StoragePolicy, aggTypes AggregationID) Policy {
-	return Policy{sp: sp, aggID: aggTypes}
+	return Policy{StoragePolicy: sp, AggregationID: aggTypes}
 }
 
 // NewPolicyFromSchema creates a new policy from a schema policy.
@@ -70,16 +70,6 @@ func NewPolicyFromSchema(p *schema.Policy) (Policy, error) {
 
 }
 
-// StoragePolicy return the storage policy.
-func (p Policy) StoragePolicy() StoragePolicy {
-	return p.sp
-}
-
-// AggregationID return the AggregationID.
-func (p Policy) AggregationID() AggregationID {
-	return p.aggID
-}
-
 // UnmarshalYAML unmarshals a policy value from a string.
 func (p *Policy) UnmarshalYAML(unmarshal func(interface{}) error) error {
 	var str string
@@ -126,10 +116,10 @@ func ParsePolicy(str string) (Policy, error) {
 
 // String is the string representation of a policy.
 func (p Policy) String() string {
-	if p.aggID.IsDefault() {
-		return p.sp.String()
+	if p.AggregationID.IsDefault() {
+		return p.StoragePolicy.String()
 	}
-	return p.sp.String() + policyAggregationTypeSeparator + p.aggID.String()
+	return p.StoragePolicy.String() + policyAggregationTypeSeparator + p.AggregationID.String()
 }
 
 // NewPoliciesFromSchema creates multiple new policies from given schema policies.
@@ -155,30 +145,30 @@ func (pr ByResolutionAsc) Len() int      { return len(pr) }
 func (pr ByResolutionAsc) Swap(i, j int) { pr[i], pr[j] = pr[j], pr[i] }
 
 func (pr ByResolutionAsc) Less(i, j int) bool {
-	up1, up2 := pr[i], pr[j]
-	p1, p2 := up1.StoragePolicy(), up2.StoragePolicy()
-	rw1, rw2 := p1.Resolution().Window, p2.Resolution().Window
+	p1, p2 := pr[i], pr[j]
+	sp1, sp2 := p1.StoragePolicy, p2.StoragePolicy
+	rw1, rw2 := sp1.Resolution().Window, sp2.Resolution().Window
 	if rw1 < rw2 {
 		return true
 	}
 	if rw1 > rw2 {
 		return false
 	}
-	r1, r2 := p1.Retention(), p2.Retention()
+	r1, r2 := sp1.Retention(), sp2.Retention()
 	if r1 > r2 {
 		return true
 	}
 	if r1 < r2 {
 		return false
 	}
-	rp1, rp2 := p1.Resolution().Precision, p2.Resolution().Precision
+	rp1, rp2 := sp1.Resolution().Precision, sp2.Resolution().Precision
 	if rp1 < rp2 {
 		return true
 	}
 	if rp1 > rp2 {
 		return false
 	}
-	at1, at2 := up1.AggregationID(), up2.AggregationID()
+	at1, at2 := p1.AggregationID, p2.AggregationID
 	for k := 0; k < AggregationIDLen; k++ {
 		if at1[k] < at2[k] {
 			return true

policy/policy_test.go

@@ -133,14 +133,8 @@ func TestNewPoliciesFromSchema(t *testing.T) {
 	res, err := NewPoliciesFromSchema(input)
 	require.NoError(t, err)
 	require.Equal(t, []Policy{
-		Policy{
-			sp:    NewStoragePolicy(10*time.Second, xtime.Second, 24*time.Hour),
-			aggID: mustCompress(Mean, P999),
-		},
-		Policy{
-			sp:    NewStoragePolicy(time.Minute, xtime.Minute, 240*time.Hour),
-			aggID: mustCompress(Mean, P9999),
-		},
+		NewPolicy(NewStoragePolicy(10*time.Second, xtime.Second, 24*time.Hour), mustCompress(Mean, P999)),
+		NewPolicy(NewStoragePolicy(time.Minute, xtime.Minute, 240*time.Hour), mustCompress(Mean, P9999)),
 	}, res)
 }
 

policy/staged_policy.go

@@ -112,16 +112,17 @@ func (l PoliciesList) IsDefault() bool {
 	return len(l) == 1 && l[0].IsDefault()
 }
 
-// WithDefaultAggregation updates the PoliciesList with default aggregation types.
-func (l PoliciesList) WithDefaultAggregation() PoliciesList {
+// SetDefaultAggregation updates the PoliciesList with default aggregation types.
+// NB(cw) This function updates the PoliciesList in place
+func (l PoliciesList) SetDefaultAggregation() PoliciesList {
 	for _, sp := range l {
 		pl, ok := sp.Policies()
 		if ok {
 			continue
 		}
 		for j := range pl {
-			if !pl[j].aggID.IsDefault() {
-				sp.policies[j] = NewPolicy(pl[j].sp, DefaultAggregationID)
+			if !pl[j].AggregationID.IsDefault() {
+				sp.policies[j] = NewPolicy(pl[j].StoragePolicy, DefaultAggregationID)
 			}
 		}
 	}

policy/staged_policy_test.go

@@ -222,15 +222,15 @@ func TestPoliciesListWithDefaultAggregation(t *testing.T) {
 		},
 	)
 
-	pl2 := pl.WithDefaultAggregation()
+	pl2 := pl.SetDefaultAggregation()
 
 	require.Equal(t, pl, pl2)
 	require.Equal(t, 1, len(pl2))
 
 	require.Equal(t, int64(100), pl2[0].CutoverNanos)
 	require.Equal(t, true, pl2[0].Tombstoned)
 	for i := range pl2[0].policies {
-		require.Equal(t, NewStoragePolicy(time.Minute, xtime.Minute, 12*time.Hour), pl2[0].policies[i].sp)
-		require.Equal(t, DefaultAggregationID, pl2[0].policies[i].aggID)
+		require.Equal(t, NewStoragePolicy(time.Minute, xtime.Minute, 12*time.Hour), pl2[0].policies[i].StoragePolicy)
+		require.Equal(t, DefaultAggregationID, pl2[0].policies[i].AggregationID)
 	}
 }

policy/storage_policy.go

@@ -60,7 +60,7 @@ func NewStoragePolicy(window time.Duration, precision xtime.Unit, retention time
 	}
 }
 
-// NewStoragePolicyFromSchema creates a new unaggregated policy from a schema policy.
+// NewStoragePolicyFromSchema creates a new storage policy from a schema storage policy.
 func NewStoragePolicyFromSchema(p *schema.StoragePolicy) (StoragePolicy, error) {
 	if p == nil {
 		return DefaultStoragePolicy, errNilStoragePolicySchema
@@ -75,22 +75,22 @@ func NewStoragePolicyFromSchema(p *schema.StoragePolicy) (StoragePolicy, error)
 
 }
 
-// String is the string representation of a policy.
+// String is the string representation of a storage policy.
 func (p StoragePolicy) String() string {
 	return fmt.Sprintf("%s%s%s", p.resolution.String(), resolutionRetentionSeparator, p.retention.String())
 }
 
-// Resolution returns the resolution of the policy.
+// Resolution returns the resolution of the storage policy.
 func (p StoragePolicy) Resolution() Resolution {
 	return p.resolution
 }
 
-// Retention return the retention of the policy.
+// Retention return the retention of the storage policy.
 func (p StoragePolicy) Retention() Retention {
 	return p.retention
 }
 
-// UnmarshalYAML unmarshals a policy value from a string.
+// UnmarshalYAML unmarshals a storage policy value from a string.
 func (p *StoragePolicy) UnmarshalYAML(unmarshal func(interface{}) error) error {
 	var str string
 	if err := unmarshal(&str); err != nil {
@@ -104,7 +104,7 @@ func (p *StoragePolicy) UnmarshalYAML(unmarshal func(interface{}) error) error {
 	return nil
 }
 
-// ParseStoragePolicy parses a policy in the form of resolution:retention.
+// ParseStoragePolicy parses a storage policy in the form of resolution:retention.
 func ParseStoragePolicy(str string) (StoragePolicy, error) {
 	parts := strings.Split(str, resolutionRetentionSeparator)
 	if len(parts) != 2 {