CompressedAggregationTypes

.excludelint

@@ -1,3 +1,4 @@
 (vendor/)
 (generated/)
 (_mock.go)
+(_string.go)

generated/proto/policy.proto

@@ -33,4 +33,22 @@ message Retention {
 message Policy {
 	Resolution resolution = 1;
 	Retention retention = 2;
+	repeated AggregationType aggregation_types = 3;
+}
+
+enum AggregationType {
+  UNKNOWN = 0;
+  LAST = 1;
+  MIN = 2;
+  MAX = 3;
+  MEAN = 4;
+  MEDIAN = 5;
+  COUNT = 6;
+  SUM = 7;
+  SUMSQ = 8;
+  STDEV = 9;
+  P95 = 10;
+  P99 = 11;
+  P99POINT9 = 12;
+  P99POINT99 = 13;
 }

policy/aggregation_type.go

@@ -0,0 +1,218 @@
+// 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 policy
+
+import (
+	"fmt"
+	"strings"
+
+	"github.com/willf/bitset"
+)
+
+// Supported AggregationTypes.
+const (
+	Unknown AggregationType = iota
+	Last
+	Min
+	Max
+	Mean
+	Median
+	Count
+	Sum
+	SumSq
+	Stdev
+	P95
+	P99
+	P99Point9
+	P99Point99
+)
+
+const (
+	// NB(cw) this should be updated if new AggregationTypes were added.
+	totalAggregationTypes = 13
+
+	// CompressedSize is the length of the CompressedAggregationTypes.
+	// The CompressedSize will be 1 when totalAggregationTypes <= 63
+	CompressedSize = totalAggregationTypes/64 + 1
+)
+
+var (
+	// EmptyAggregationTypes is an empty list of AggregationTypes.
+	EmptyAggregationTypes AggregationTypes
+
+	// EmptyCompressedAggregationTypes is an empty CompressedAggregationTypes.
+	EmptyCompressedAggregationTypes CompressedAggregationTypes
+)
+
+// AggregationType defines a custom aggregation function.
+type AggregationType int
+
+// IsValid checks if an AggregationType is valid.
+func (a AggregationType) IsValid() bool {
+	return a > 0 && a <= totalAggregationTypes
+}
+
+// IsValidForGauge if an AggregationType is valid for Gauge
+func (a AggregationType) IsValidForGauge() bool {
+	switch a {
+	case Last, Min, Max, Mean, Count, Sum, SumSq, Stdev:
+		return true
+	default:
+		return false
+	}
+}
+
+// IsValidForCounter if an AggregationType is valid for Counter
+func (a AggregationType) IsValidForCounter() bool {
+	switch a {
+	case Min, Max, Mean, Count, Sum, SumSq, Stdev:
+		return true
+	default:
+		return false
+	}
+}
+
+// IsValidForTimer if an AggregationType is valid for Timer
+func (a AggregationType) IsValidForTimer() bool {
+	switch a {
+	case Last:
+		return false
+	default:
+		return true
+	}
+}
+
+// CompressedAggregationTypes is a compressed AggregationTypes.
+type CompressedAggregationTypes [CompressedSize]int64
+
+// IsDefault returns if the CompressedAggregationTypes equals the default aggregation type
+func (compressed CompressedAggregationTypes) IsDefault() bool {
+	if compressed == EmptyCompressedAggregationTypes {
+		return true
+	}
+	return false
+}
+
+// Merge returns the result of merging another CompressedAggregationTypes, with an indicater whether
+// any new ggregation type were found in the other CompressedAggregationTypes
+func (compressed CompressedAggregationTypes) Merge(other CompressedAggregationTypes) (CompressedAggregationTypes, bool) {
+	var merged bool
+	for i, code := range compressed {
+		otherCode := other[i]
+		if otherCode != 0 && code != otherCode {
+			merged = true
+			compressed[i] = code | otherCode
+		}
+	}
+	return compressed, merged
+}
+
+// String for debugging
+func (compressed CompressedAggregationTypes) String() string {
+	aggTypes, err := NewAggregationTypeDecompressor().Decompress(compressed)
+	if err != nil {
+		return fmt.Sprintf("[invalid CompressedAggregationTypes: %v]", err)
+	}
+
+	return aggTypes.String()
+}
+
+// AggregationTypes is a list of AggregationTypes.
+type AggregationTypes []AggregationType
+
+func (aggTypes AggregationTypes) String() string {
+	if len(aggTypes) == 0 {
+		return ""
+	}
+
+	parts := make([]string, len(aggTypes))
+	for i, aggType := range aggTypes {
+		parts[i] = aggType.String()
+	}
+	return "[" + strings.Join(parts, ",") + "]"
+}
+
+// AggregationTypeCompressor can compress AggregationTypes.
+type AggregationTypeCompressor interface {
+	Compress(aggTypes AggregationTypes) (CompressedAggregationTypes, error)
+}
+
+// AggregationTypeDecompressor can decompress AggregationTypes.
+type AggregationTypeDecompressor interface {
+	Decompress(code CompressedAggregationTypes) (AggregationTypes, error)
+}
+
+type aggregationCompresser struct {
+	bs *bitset.BitSet
+}
+
+// NewAggregationTypeCompressor returns a new AggregationTypeCompressor.
+func NewAggregationTypeCompressor() AggregationTypeCompressor {
+	// NB(cw): If we start to support more than 64 types, the library will
+	// expand the underlying word list itself.
+	return &aggregationCompresser{bs: bitset.New(CompressedSize)}
+}
+
+func (c *aggregationCompresser) Compress(aggTypes AggregationTypes) (CompressedAggregationTypes, error) {
+	c.bs.ClearAll()
+	for _, aggType := range aggTypes {
+		if !aggType.IsValid() {
+			return EmptyCompressedAggregationTypes, fmt.Errorf("could not compress invalid AggregationType %v", aggType)
+		}
+		c.bs.Set(uint(aggType))
+	}
+	codes := c.bs.Bytes()
+	var res CompressedAggregationTypes
+	for i := 0; i < CompressedSize && i < len(codes); i++ {
+		res[i] = int64(codes[i])
+	}
+	return res, nil
+}
+
+type aggregationDecompresser struct {
+	res []AggregationType
+}
+
+// NewAggregationTypeDecompressor returns a new AggregationTypeDecompressor.
+func NewAggregationTypeDecompressor() AggregationTypeDecompressor {
+	return &aggregationDecompresser{res: make([]AggregationType, 0, totalAggregationTypes)}
+}
+
+func (c *aggregationDecompresser) Decompress(compressed CompressedAggregationTypes) (AggregationTypes, error) {
+	codes := make([]uint64, CompressedSize)
+	for i := range compressed {
+		codes[i] = uint64(compressed[i])
+	}
+	bs := bitset.From([]uint64(codes))
+
+	c.res = c.res[:0]
+	for i, e := bs.NextSet(0); e; i, e = bs.NextSet(i + 1) {
+		aggType := AggregationType(i)
+		if !aggType.IsValid() {
+			return EmptyAggregationTypes, fmt.Errorf("could not decompress invalid AggregationType %v", i)
+		}
+		c.res = append(c.res, aggType)
+	}
+
+	res := make(AggregationTypes, len(c.res))
+	copy(res, c.res)
+	return res, nil
+}